factor_engine/single_factor_backtest.py

"""Single-factor IC analysis and strategy backtest."""

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

import pandas as pd

from config.settings import load_settings
from factor_engine.factor_evaluation import daily_rank_ic, evaluate_factor_panel, ic_summary, quantile_spread
from factor_engine.formula_registry import (
    compute_factor,
    factor_series_to_panel,
    get_factor_spec,
    list_factor_specs,
    load_label_panel,
    registry_output_dir,
)
from strategies.runner import run_strategy_backtest, save_backtest_result


def build_factor_label_panel(
    factor_name: str,
    segment: str = "test",
    start_time: str | None = None,
    end_time: str | None = None,
) -> pd.DataFrame:
    settings = load_settings()
    if start_time is None or end_time is None:
        seg = settings.segments.get(segment, settings.segments["test"])
        start_time = start_time or seg[0]
        end_time = end_time or seg[1]

    spec = get_factor_spec(factor_name)
    factor_s = compute_factor(factor_name, start_time=start_time, end_time=end_time, cache=True)
    factor_panel = factor_series_to_panel(factor_s, factor_name)
    label_panel = load_label_panel(start_time=start_time, end_time=end_time, label_expr=spec.label_expr)

    merged = factor_panel.merge(label_panel, on=["date", "symbol"], how="inner")
    merged = merged.rename(columns={factor_name: "factor"})
    merged = merged.dropna(subset=["factor", "label"])
    return merged


def analyze_single_factor(
    factor_name: str,
    segment: str = "test",
    start_time: str | None = None,
    end_time: str | None = None,
) -> dict[str, Any]:
    panel = build_factor_label_panel(factor_name, segment=segment, start_time=start_time, end_time=end_time)
    metrics = evaluate_factor_panel(panel)
    ic_series = daily_rank_ic(panel)
    spread = quantile_spread(panel)

    spec = get_factor_spec(factor_name)
    return {
        "factor_name": factor_name,
        "expression": spec.expression,
        "description": spec.description,
        "segment": segment,
        "n_obs": len(panel),
        "metrics": metrics,
        "ic_series": ic_series,
        "quantile_spread": spread,
    }


def backtest_single_factor(
    factor_name: str,
    strategy_name: str = "topk_dropout",
    strategy_kwargs: dict[str, Any] | None = None,
    segment: str = "test",
    start_time: str | None = None,
    end_time: str | None = None,
    output_dir: Path | None = None,
) -> dict[str, Any]:
    settings = load_settings()
    if start_time is None or end_time is None:
        seg = settings.segments.get(segment, settings.segments["test"])
        start_time = start_time or seg[0]
        end_time = end_time or seg[1]

    analysis = analyze_single_factor(factor_name, segment=segment, start_time=start_time, end_time=end_time)

    signal_source = {
        "type": "factor_registry",
        "name": factor_name,
        "start_time": start_time,
        "end_time": end_time,
    }
    bt_result = run_strategy_backtest(
        strategy_name=strategy_name,
        signal_source=signal_source,
        strategy_kwargs=strategy_kwargs,
        start_time=start_time,
        end_time=end_time,
    )

    out_dir = output_dir or settings.output_root / "factors" / "single_backtest" / factor_name
    save_backtest_result(bt_result, out_dir)

    ic_path = registry_output_dir() / "ic_summary.csv"
    ic_row = {
        "factor_name": factor_name,
        "expression": analysis["expression"],
        **analysis["metrics"],
        "strategy": strategy_name,
        "segment": segment,
    }
    if ic_path.exists():
        summary_df = pd.read_csv(ic_path)
        summary_df = summary_df[summary_df["factor_name"] != factor_name]
        summary_df = pd.concat([summary_df, pd.DataFrame([ic_row])], ignore_index=True)
    else:
        summary_df = pd.DataFrame([ic_row])
    summary_df.to_csv(ic_path, index=False)

    result = {
        "factor_name": factor_name,
        "analysis": {k: v for k, v in analysis.items() if k not in ("ic_series", "quantile_spread")},
        "backtest": {
            "strategy": strategy_name,
            "signal_stats": bt_result.signal_stats,
            "risk": bt_result.risk.to_dict() if not bt_result.risk.empty else {},
            "output_dir": str(out_dir),
        },
    }

    with open(out_dir / "single_factor_report.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2, default=str)

    return result


def run_all_single_factor_backtests(
    strategy_name: str = "topk_dropout",
    segment: str = "test",
    enabled_only: bool = True,
) -> pd.DataFrame:
    rows = []
    for spec in list_factor_specs(enabled_only=enabled_only):
        print(f"\n=== Single factor backtest: {spec.name} ===")
        try:
            res = backtest_single_factor(spec.name, strategy_name=strategy_name, segment=segment)
            row = {"factor_name": spec.name, **res["analysis"]["metrics"]}
            if res["backtest"]["risk"]:
                row["ann_return"] = res["backtest"]["risk"].get("annualized_return")
                row["max_drawdown"] = res["backtest"]["risk"].get("max_drawdown")
            rows.append(row)
            print(f"  ICIR={row.get('icir', 'NA'):.4f}")
        except Exception as exc:
            print(f"  FAILED: {exc}")
            rows.append({"factor_name": spec.name, "error": str(exc)})

    df = pd.DataFrame(rows)
    out = registry_output_dir() / "all_single_factor_summary.csv"
    df.to_csv(out, index=False)
    print(f"\nSummary saved: {out}")
    return df