File size: 2,460 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 | """Qlib-integrated backtest engine."""
from __future__ import annotations
import pandas as pd
from backtest.performance_metrics import summarize_returns
from backtest.portfolio_construction import top_k_equal_weight
from backtest.transaction_cost import apply_transaction_cost
from data_pipeline.init_qlib import init_qlib
from qlib.contrib.evaluate import risk_analysis
from qlib.contrib.strategy import TopkDropoutStrategy
from qlib.backtest import backtest as qlib_backtest
def run_qlib_topk_backtest(
pred_score: pd.Series,
topk: int = 30,
n_drop: int = 5,
start_time: str = "2024-01-01",
end_time: str = "2025-12-31",
config_path: str | None = None,
) -> dict:
"""
Run qlib TopkDropout backtest on model prediction scores.
pred_score: MultiIndex (instrument, datetime) Series.
"""
init_qlib(config_path)
strategy = TopkDropoutStrategy(
signal=pred_score,
topk=topk,
n_drop=n_drop,
)
executor_config = {
"class": "SimulatorExecutor",
"module_path": "qlib.backtest.executor",
"kwargs": {"time_per_step": "day", "generate_portfolio_metrics": True},
}
portfolio_metric_dict, positions = qlib_backtest(
start_time=start_time,
end_time=end_time,
strategy=strategy,
executor=executor_config,
account=100_000_000,
benchmark="SH000300",
)
freq_key = next(iter(portfolio_metric_dict.keys()))
report, pos = portfolio_metric_dict[freq_key]
analysis = risk_analysis(report["return"])
return {"report": report, "positions": pos, "risk": analysis}
def simple_long_only_backtest(
panel: pd.DataFrame,
score_col: str = "score",
ret_col: str = "label",
top_k: int = 30,
cost_rate: float = 0.0015,
) -> dict:
"""Lightweight backtest without full qlib exchange simulator."""
daily_rets = []
prev_holdings = set()
for dt, group in panel.groupby("date"):
top = group.nlargest(top_k, score_col)
holdings = set(top["symbol"])
turnover = len(holdings.symmetric_difference(prev_holdings)) / max(2 * top_k, 1)
gross = top[ret_col].mean()
net = gross - apply_transaction_cost(turnover, cost_rate)
daily_rets.append({"date": dt, "return": net})
prev_holdings = holdings
ret_series = pd.Series({r["date"]: r["return"] for r in daily_rets}).sort_index()
return summarize_returns(ret_series)
|