Blame for timeout_risk_model.py

"""Small timeout-risk model for Huaxiang ChaPanda orders.

The first dataset only contains delayed orders, so the model supports two modes:

* labeled_empirical_bayes: used when both timeout and non-timeout samples exist.
* positive_only_time_density: used when only timeout samples exist. This is a
  relative risk estimate by order time, not a calibrated probability.
"""

from __future__ import annotations

import csv
import json
import math
from collections import Counter, defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable

MODEL_VERSION = 1
CSV_FIELDS = [
    "product_name",
    "order_time",
    "timeout",
    "is_timeout",
    "expected_latest_time",
    "actual_arrival_time",
    "submitted_at",
    "notes",
]


@dataclass(frozen=True)
class TrainingRecord:
    product_name: str
    order_time: datetime
    is_timeout: bool
    source: str


def parse_datetime(value: Any) -> datetime | None:
    if value is None:
        return None
    text = str(value).strip()
    if not text:
        return None

    normalized = text.replace("/", "-").replace("T", " ")
    if len(normalized) == 16:
        normalized = f"{normalized}:00"

    formats = (
        "%Y-%m-%d %H:%M:%S",
        "%Y-%m-%d %H:%M",
        "%Y-%m-%d",
    )
    for fmt in formats:
        try:
            return datetime.strptime(normalized, fmt)
        except ValueError:
            pass

    try:
        return datetime.fromisoformat(normalized)
    except ValueError:
        return None


def normalize_row(row: dict[str, Any]) -> dict[str, str]:
    return {str(key).strip(): str(value).strip() for key, value in row.items() if key is not None}


def coerce_bool(value: Any) -> bool | None:
    if value is None:
        return None
    text = str(value).strip().lower()
    if not text:
        return None
    true_values = {"1", "true", "yes", "y", "timeout", "late", "超时", "是", "yes_timeout"}
    false_values = {"0", "false", "no", "n", "on_time", "not_timeout", "未超时", "否"}
    if text in true_values:
        return True
    if text in false_values:
        return False
    if "未超时" in text or "准时" in text:
        return False
    if "超时" in text or "延迟" in text:
        return True
    return None


def infer_timeout_label(row: dict[str, str], forced_timeout: bool | None = None) -> bool | None:
    if forced_timeout is not None:
        return forced_timeout

    for key in ("is_timeout", "timeout", "timeout_status", "status"):
        parsed = coerce_bool(row.get(key))
        if parsed is not None:
            return parsed

    expected = parse_datetime(row.get("expected_latest_time"))
    actual = parse_datetime(row.get("actual_arrival_time"))
    if expected and actual:
        return actual > expected
    return None


def read_csv_records(path: Path, forced_timeout: bool | None = None) -> list[TrainingRecord]:
    if not path.exists():
        return []

    records: list[TrainingRecord] = []
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        reader = csv.DictReader(handle)
        for raw_row in reader:
            row = normalize_row(raw_row)
            order_time = parse_datetime(row.get("order_time"))
            is_timeout = infer_timeout_label(row, forced_timeout=forced_timeout)
            if not order_time or is_timeout is None:
                continue

            product_name = row.get("product_name", "").strip() or "unknown"
            records.append(
                TrainingRecord(
                    product_name=product_name,
                    order_time=order_time,
                    is_timeout=is_timeout,
                    source=str(path),
                )
            )
    return records


def load_dataset_records(dataset_dir: str | Path) -> list[TrainingRecord]:
    dataset_path = Path(dataset_dir)
    records_path = dataset_path / "records.csv"
    if records_path.exists():
        records = read_csv_records(records_path, forced_timeout=None)
    else:
        records = read_csv_records(dataset_path / "timeout_records.csv", forced_timeout=True)

    tmp_dir = dataset_path / "tmp"
    if tmp_dir.exists():
        for tmp_file in sorted(tmp_dir.glob("tmp_records_*.csv")):
            records.extend(read_csv_records(tmp_file, forced_timeout=None))

    return records


def _time_bucket(dt: datetime) -> str:
    hour = dt.hour
    if 6 <= hour < 11:
        return "morning"
    if 11 <= hour < 14:
        return "lunch"
    if 14 <= hour < 17:
        return "afternoon"
    if 17 <= hour < 20:
        return "dinner"
    if 20 <= hour < 24:
        return "night"
    return "late_night"


def _sigmoid(value: float) -> float:
    return 1.0 / (1.0 + math.exp(-value))


def _logit(probability: float) -> float:
    clipped = min(max(probability, 0.02), 0.98)
    return math.log(clipped / (1.0 - clipped))


def _clip_probability(value: float) -> float:
    return round(min(max(value, 0.02), 0.98), 4)


def _empirical_probabilities(
    records: list[TrainingRecord],
    key_fn: Callable[[TrainingRecord], str],
    base_probability: float,
    prior_strength: float,
) -> dict[str, dict[str, float | int]]:
    grouped: dict[str, list[TrainingRecord]] = defaultdict(list)
    for record in records:
        grouped[key_fn(record)].append(record)

    stats: dict[str, dict[str, float | int]] = {}
    for key, items in grouped.items():
        total = len(items)
        timeouts = sum(1 for item in items if item.is_timeout)
        probability = (timeouts + prior_strength * base_probability) / (total + prior_strength)
        stats[str(key)] = {
            "total": total,
            "timeouts": timeouts,
            "probability": _clip_probability(probability),
        }
    return stats


def _positive_density_probabilities(
    records: list[TrainingRecord],
    key_fn: Callable[[TrainingRecord], str],
    keys: Iterable[str],
    low: float,
    high: float,
) -> dict[str, dict[str, float | int]]:
    counts = Counter(key_fn(record) for record in records if record.is_timeout)
    max_count = max(counts.values(), default=0)
    output: dict[str, dict[str, float | int]] = {}
    for key in keys:
        count = counts.get(key, 0)
        density = count / max_count if max_count else 0.0
        probability = low + (high - low) * density
        output[str(key)] = {
            "positive_count": count,
            "probability": _clip_probability(probability),
        }
    return output


def _smoothed_hour_density(records: list[TrainingRecord]) -> dict[str, dict[str, float | int]]:
    raw_counts = Counter(record.order_time.hour for record in records if record.is_timeout)
    smoothed: dict[int, float] = {}
    for hour in range(24):
        smoothed[hour] = (
            raw_counts.get(hour, 0)
            + 0.45 * raw_counts.get((hour - 1) % 24, 0)
            + 0.45 * raw_counts.get((hour + 1) % 24, 0)
            + 0.15 * raw_counts.get((hour - 2) % 24, 0)
            + 0.15 * raw_counts.get((hour + 2) % 24, 0)
        )
    max_score = max(smoothed.values(), default=0.0)

    output: dict[str, dict[str, float | int]] = {}
    for hour in range(24):
        density = smoothed[hour] / max_score if max_score else 0.0
        probability = 0.12 + 0.68 * density
        output[str(hour)] = {
            "positive_count": raw_counts.get(hour, 0),
            "smoothed_density": round(density, 4),
            "probability": _clip_probability(probability),
        }
    return output


def train_timeout_model(records: list[TrainingRecord]) -> dict[str, Any]:
    total = len(records)
    positives = sum(1 for record in records if record.is_timeout)
    negatives = total - positives
    generated_at = datetime.now(timezone.utc).isoformat()

    if total == 0:
        return {
            "model_version": MODEL_VERSION,
            "mode": "empty_default",
            "generated_at": generated_at,
            "base_probability": 0.35,
            "hour_probabilities": {str(hour): {"probability": 0.35, "total": 0, "timeouts": 0} for hour in range(24)},
            "weekday_probabilities": {str(day): {"probability": 0.35, "total": 0, "timeouts": 0} for day in range(7)},
            "time_bucket_probabilities": {},
            "product_probabilities": {},
            "metadata": {
                "training_records": 0,
                "timeout_records": 0,
                "non_timeout_records": 0,
                "calibration": "No valid records were found. Default probability is a placeholder.",
            },
        }

    if positives > 0 and negatives > 0:
        base_probability = (positives + 2.0) / (total + 4.0)
        mode = "labeled_empirical_bayes"
        calibration = "Calibrated from user-labeled timeout and non-timeout records with Bayesian smoothing."
        hour_probabilities = _empirical_probabilities(records, lambda r: str(r.order_time.hour), base_probability, 6.0)
        weekday_probabilities = _empirical_probabilities(records, lambda r: str(r.order_time.weekday()), base_probability, 8.0)
        time_bucket_probabilities = _empirical_probabilities(records, lambda r: _time_bucket(r.order_time), base_probability, 5.0)
        product_probabilities = _empirical_probabilities(records, lambda r: r.product_name, base_probability, 8.0)
    else:
        base_probability = 0.35
        mode = "positive_only_time_density"
        calibration = (
            "Only timeout records were available. Output is a relative time-risk estimate, "
            "not a fully calibrated timeout probability."
        )
        positive_records = [record for record in records if record.is_timeout]
        hour_probabilities = _smoothed_hour_density(positive_records)
        weekday_probabilities = _positive_density_probabilities(
            positive_records,
            lambda r: str(r.order_time.weekday()),
            [str(day) for day in range(7)],
            low=0.18,
            high=0.62,
        )
        time_bucket_probabilities = _positive_density_probabilities(
            positive_records,
            lambda r: _time_bucket(r.order_time),
            ["morning", "lunch", "afternoon", "dinner", "night", "late_night"],
            low=0.16,
            high=0.70,
        )
        product_probabilities = {}

    return {
        "model_version": MODEL_VERSION,
        "mode": mode,
        "generated_at": generated_at,
        "base_probability": _clip_probability(base_probability),
        "hour_probabilities": hour_probabilities,
        "weekday_probabilities": weekday_probabilities,
        "time_bucket_probabilities": time_bucket_probabilities,
        "product_probabilities": product_probabilities,
        "metadata": {
            "training_records": total,
            "timeout_records": positives,
            "non_timeout_records": negatives,
            "calibration": calibration,
            "features": ["order hour", "weekday", "time bucket", "optional product name"],
        },
    }


def load_model(path: str | Path) -> dict[str, Any]:
    with Path(path).open("r", encoding="utf-8") as handle:
        return json.load(handle)


def save_model(model: dict[str, Any], path: str | Path) -> None:
    output_path = Path(path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with output_path.open("w", encoding="utf-8") as handle:
        json.dump(model, handle, ensure_ascii=False, indent=2, sort_keys=True)
        handle.write("\n")


def predict_timeout_probability(
    model: dict[str, Any],
    order_time: str | datetime,
    product_name: str = "",
) -> dict[str, Any]:
    dt = order_time if isinstance(order_time, datetime) else parse_datetime(order_time)
    if dt is None:
        raise ValueError("order_time must look like '2026-05-24 18:30'.")

    base = float(model.get("base_probability", 0.35))
    mode = model.get("mode", "unknown")

    def component_probability(section: str, key: str, fallback: float) -> float:
        value = model.get(section, {}).get(str(key), {})
        try:
            return float(value.get("probability", fallback))
        except (TypeError, ValueError):
            return fallback

    hour_probability = component_probability("hour_probabilities", str(dt.hour), base)
    weekday_probability = component_probability("weekday_probabilities", str(dt.weekday()), base)
    bucket_probability = component_probability("time_bucket_probabilities", _time_bucket(dt), base)
    product_probability = None
    if product_name:
        product_probability = component_probability("product_probabilities", product_name.strip(), base)

    if mode == "labeled_empirical_bayes":
        combined_logit = _logit(base)
        combined_logit += 0.58 * (_logit(hour_probability) - _logit(base))
        combined_logit += 0.18 * (_logit(weekday_probability) - _logit(base))
        combined_logit += 0.16 * (_logit(bucket_probability) - _logit(base))
        if product_probability is not None:
            combined_logit += 0.08 * (_logit(product_probability) - _logit(base))
        probability = _sigmoid(combined_logit)
    else:
        combined_logit = _logit(base)
        combined_logit += 0.70 * (_logit(hour_probability) - _logit(base))
        combined_logit += 0.18 * (_logit(bucket_probability) - _logit(base))
        combined_logit += 0.12 * (_logit(weekday_probability) - _logit(base))
        probability = _sigmoid(combined_logit)

    clipped = _clip_probability(probability)
    if clipped >= 0.65:
        level = "high"
    elif clipped >= 0.4:
        level = "medium"
    else:
        level = "low"

    return {
        "probability": clipped,
        "risk_level": level,
        "mode": mode,
        "order_time": dt.strftime("%Y-%m-%d %H:%M"),
        "components": {
            "base": base,
            "hour": hour_probability,
            "weekday": weekday_probability,
            "time_bucket": bucket_probability,
            "product": product_probability,
        },
        "calibration": model.get("metadata", {}).get("calibration", ""),
    }