File size: 5,635 Bytes
590a501 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 | """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
|