#!/usr/bin/env python
"""Strict walk-forward backtest of the weighted-speed forecast."""

from __future__ import annotations

import argparse
import csv
import json
from datetime import date, datetime, time, timedelta
from decimal import Decimal
from pathlib import Path
from typing import Any

import pandas as pd

from odps_source import build_odps_client
from revenue_forecast import (
    RevenueObservation,
    RevenueTrendWindow,
)
from revenue_forecast_config import RevenueForecastConfig
from revenue_forecast_source import fetch_interval_revenue_series
from revenue_speed_forecast import (
    aggregate_speed_parameters,
    build_speed_samples,
    calculate_speed_forecast,
    decimal_quantile,
)
from run_once import load_environment


OUTPUT_DIR = Path(__file__).resolve().parent / "outputs"


def _parse_date(value: str) -> date:
    return datetime.strptime(value, "%Y%m%d").date()


def _date_range(start: date, end: date) -> tuple[date, ...]:
    return tuple(
        start + timedelta(days=offset)
        for offset in range((end - start).days + 1)
    )


def _pct_error(predicted: Decimal, actual: Decimal) -> Decimal:
    return (predicted - actual) / actual * Decimal("100")


def _mean(values: list[Decimal]) -> Decimal | None:
    return sum(values, Decimal("0")) / len(values) if values else None


def _model_summary(rows: list[dict[str, Any]]) -> dict[str, Any]:
    errors = [abs(row["error_pct"]) for row in rows]
    signed = [row["error_pct"] for row in rows]
    jumps = [
        row["jump_pct"]
        for row in rows
        if row["jump_pct"] is not None
    ]
    return {
        "rows": len(rows),
        "mape_pct": _mean(errors),
        "signed_bias_pct": _mean(signed),
        "p90_abs_error_pct": (
            decimal_quantile(errors, Decimal("0.9")) if errors else None
        ),
        "over_20pct_rate_pct": (
            Decimal(sum(error > 20 for error in errors))
            / Decimal(len(errors))
            * Decimal("100")
            if errors
            else None
        ),
        "mean_prediction_jump_pct": _mean(jumps),
    }


def summarize(rows: list[dict[str, Any]]) -> dict[str, Any]:
    result: dict[str, Any] = {"rows": len(rows), "windows": {}}
    for label, start_at in (
        ("from_07", "07:00"),
        ("from_09", "09:00"),
        ("from_12", "12:00"),
        ("from_15", "15:00"),
    ):
        selected = [row for row in rows if row["time"] >= start_at]
        result["windows"][label] = {
            "weighted_speed": _model_summary(selected),
            "interval_hit_rate_pct": (
                Decimal(sum(row["interval_hit"] for row in selected))
                / Decimal(len(selected))
                * Decimal("100")
                if selected
                else None
            ),
        }
    result["by_date"] = {}
    for data_date in sorted({row["date"] for row in rows}):
        selected = [row for row in rows if row["date"] == data_date]
        result["by_date"][data_date] = {
            "weighted_speed": _model_summary(selected),
        }
    return result


def _json_default(value: Any) -> Any:
    if isinstance(value, Decimal):
        return str(value)
    raise TypeError(f"Unsupported JSON value: {type(value)!r}")


def run_backtest(
    *,
    series: dict[date, list[tuple[datetime, Decimal]]],
    prediction_dates: tuple[date, ...],
    config: RevenueForecastConfig,
    start_at: time,
    stop_at: time,
) -> list[dict[str, Any]]:
    all_samples = []
    samples_by_date_time = {}
    for data_date, intervals in series.items():
        samples = build_speed_samples(
            data_date=data_date,
            intervals=intervals,
            parameter_version=config.speed_parameter_version,
            latest_weight=config.speed_latest_weight,
            start_time=start_at,
            stop_time=stop_at,
        )
        all_samples.extend(samples)
        for sample in samples:
            samples_by_date_time[(data_date, sample.report_time.time())] = sample

    rows: list[dict[str, Any]] = []
    previous_predictions: dict[date, Decimal] = {}
    for prediction_date in prediction_dates:
        current_intervals = series.get(prediction_date, [])
        if len(current_intervals) != 96:
            continue
        actual = sum((value for _, value in current_intervals), Decimal("0"))
        cumulative = Decimal("0")
        trend_windows: list[RevenueTrendWindow] = []
        for window_start, revenue in current_intervals:
            report_time = window_start + timedelta(minutes=15)
            cumulative += revenue
            trend_windows.append(
                RevenueTrendWindow(
                    report_time=report_time,
                    today_revenue=revenue,
                )
            )
            if not start_at <= report_time.time() <= stop_at:
                continue
            current_sample = samples_by_date_time.get(
                (prediction_date, report_time.time())
            )
            if current_sample is None:
                continue
            training_samples = [
                item
                for item in all_samples
                if item.data_date < prediction_date
                and item.report_time.time() == report_time.time()
            ]
            parameters = aggregate_speed_parameters(training_samples)
            if not parameters:
                continue
            parameter = parameters[0]
            observation = RevenueObservation(
                partition=prediction_date.strftime("%Y%m%d"),
                report_time=report_time,
                today_revenue=cumulative,
            )
            speed = calculate_speed_forecast(
                observation,
                trend_windows[-3:],
                parameter,
                forecast_version=config.forecast_version,
                parameter_version=config.speed_parameter_version,
                target_cost_ratio=config.target_cost_ratio,
                minimum_samples=1,
            )
            time_text = report_time.strftime("%H:%M")
            row = {
                "date": prediction_date.strftime("%Y%m%d"),
                "time": time_text,
                "training_start": parameter.training_start.strftime("%Y%m%d"),
                "training_end": parameter.training_end.strftime("%Y%m%d"),
                "training_samples": parameter.sample_count,
                "actual_revenue": actual,
                "current_cumulative": cumulative,
                "latest_15m_revenue": revenue,
                "previous_15m_revenue": trend_windows[-2].today_revenue,
                "weighted_speed": speed.weighted_speed,
                "remaining_multiplier_p10": parameter.multiplier_p10,
                "remaining_multiplier_p50": parameter.multiplier_p50,
                "remaining_multiplier_p90": parameter.multiplier_p90,
                "forecast_p10": speed.forecast_lower,
                "forecast_p50": speed.forecast_revenue,
                "forecast_p90": speed.forecast_upper,
                "error_pct": _pct_error(speed.forecast_revenue, actual),
                "interval_hit": (
                    speed.forecast_lower <= actual <= speed.forecast_upper
                ),
            }
            previous = previous_predictions.get(prediction_date)
            row["jump_pct"] = (
                abs(speed.forecast_revenue - previous)
                / previous
                * Decimal("100")
                if previous is not None and previous > 0
                else None
            )
            previous_predictions[prediction_date] = speed.forecast_revenue
            rows.append(row)
    return rows


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--start-date", required=True)
    parser.add_argument("--end-date", required=True)
    parser.add_argument("--history-start", required=True)
    parser.add_argument("--start-time", default="07:00")
    parser.add_argument("--stop-time", default="22:30")
    parser.add_argument("--output-dir", type=Path, default=OUTPUT_DIR)
    return parser.parse_args()


def main() -> int:
    load_environment()
    args = parse_args()
    start_date = _parse_date(args.start_date)
    end_date = _parse_date(args.end_date)
    history_start = _parse_date(args.history_start)
    if not history_start < start_date <= end_date:
        raise ValueError("history-start must be before the prediction range")
    dates = _date_range(history_start, end_date)
    series = fetch_interval_revenue_series(build_odps_client(), dates)
    rows = run_backtest(
        series=series,
        prediction_dates=_date_range(start_date, end_date),
        config=RevenueForecastConfig.from_env(),
        start_at=datetime.strptime(args.start_time, "%H:%M").time(),
        stop_at=datetime.strptime(args.stop_time, "%H:%M").time(),
    )
    if not rows:
        raise RuntimeError("No strict walk-forward rows produced")

    args.output_dir.mkdir(parents=True, exist_ok=True)
    suffix = f"{args.start_date}_{args.end_date}"
    csv_path = args.output_dir / f"revenue_speed_backtest_{suffix}.csv"
    json_path = args.output_dir / f"revenue_speed_backtest_{suffix}.json"
    xlsx_path = args.output_dir / f"revenue_speed_backtest_{suffix}.xlsx"
    with csv_path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)
    summary = summarize(rows)
    json_path.write_text(
        json.dumps(summary, ensure_ascii=False, indent=2, default=_json_default),
        encoding="utf-8",
    )
    pd.DataFrame(rows).to_excel(xlsx_path, index=False)
    print(
        json.dumps(
            {
                "summary": summary,
                "csv": str(csv_path),
                "json": str(json_path),
                "xlsx": str(xlsx_path),
            },
            ensure_ascii=False,
            default=_json_default,
        )
    )
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
