| """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", ""), |
| } |
|
|
