backtest/backtest_engine.py

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