(
dataset: ResearchDataset,
*,
config: dict[str, Any] | None = None,
)
| 99 | |
| 100 | |
| 101 | def run_flywheel_iteration( |
| 102 | dataset: ResearchDataset, |
| 103 | *, |
| 104 | config: dict[str, Any] | None = None, |
| 105 | ) -> dict[str, Any]: |
| 106 | cfg = { |
| 107 | "cusum_threshold": 0.001, |
| 108 | "num_classes": 2, |
| 109 | "step_size": 0.1, |
| 110 | "risk_free_rate": 0.0, |
| 111 | "confidence_level": 0.05, |
| 112 | "commission_bps": 1.5, |
| 113 | "spread_bps": 2.0, |
| 114 | "slippage_vol_mult": 8.0, |
| 115 | "min_net_sharpe": 0.30, |
| 116 | "min_realized_sharpe": 0.25, |
| 117 | } |
| 118 | if config: |
| 119 | cfg.update(config) |
| 120 | |
| 121 | out = pipeline.run_mid_frequency_pipeline_frames( |
| 122 | timestamps=dataset.timestamps, |
| 123 | close=dataset.close, |
| 124 | model_probabilities=dataset.model_probabilities, |
| 125 | model_sides=dataset.model_sides, |
| 126 | asset_prices=dataset.asset_prices, |
| 127 | asset_names=dataset.asset_names, |
| 128 | cusum_threshold=float(cfg["cusum_threshold"]), |
| 129 | num_classes=int(cfg["num_classes"]), |
| 130 | step_size=float(cfg["step_size"]), |
| 131 | risk_free_rate=float(cfg["risk_free_rate"]), |
| 132 | confidence_level=float(cfg["confidence_level"]), |
| 133 | ) |
| 134 | |
| 135 | backtest = out["frames"]["backtest"] |
| 136 | strategy_returns = backtest["returns"].to_list() |
| 137 | positions = backtest["position"].to_list() |
| 138 | |
| 139 | turnover = _turnover(positions) |
| 140 | realized_vol = _annualized_vol(strategy_returns) |
| 141 | cost_per_turn = ( |
| 142 | float(cfg["commission_bps"]) * 1e-4 |
| 143 | + float(cfg["spread_bps"]) * 1e-4 |
| 144 | + float(cfg["slippage_vol_mult"]) * realized_vol * 1e-3 |
| 145 | ) |
| 146 | total_cost = turnover * cost_per_turn |
| 147 | gross_total_return = backtest["equity"][-1] - 1.0 |
| 148 | net_total_return = gross_total_return - total_cost |
| 149 | |
| 150 | bars = len(strategy_returns) |
| 151 | annualizer = (252.0 * 390.0 / max(bars, 1)) ** 0.5 |
| 152 | mean_r = sum(strategy_returns) / max(bars, 1) |
| 153 | std_r = _std(strategy_returns) |
| 154 | net_sharpe = (mean_r / std_r) * annualizer if std_r > 0 else 0.0 |
| 155 | |
| 156 | promotion = { |
| 157 | "passed_realized_sharpe": out["risk"]["realized_sharpe"] >= float(cfg["min_realized_sharpe"]), |
| 158 | "passed_net_sharpe": net_sharpe >= float(cfg["min_net_sharpe"]), |
no test coverage detected