Download crypto_eval/engine2.py from Avi-L/crypto-quant-eval: direct link, hf CLI and curl.
- Browser
- Download file 3.49 kB
-
https://huggingface.co/Avi-L/crypto-quant-eval/resolve/main/crypto_eval/engine2.py
- Command line
-
hf download hf://Avi-L/crypto-quant-eval/crypto_eval/engine2.py
-
curl -L -o engine2.py https://huggingface.co/Avi-L/crypto-quant-eval/resolve/main/crypto_eval/engine2.py
3.49 kB
| """Daily multi-asset backtest engine for the panel sweep. | |
| Simpler than the 1h engine: the panel carries close prices only. | |
| Per asset: sub-capital = initial_capital / n_assets, positions in [-1, 1] | |
| of that sub-capital. Portfolio return = equal-weight mean over the assets | |
| LISTED on each date (pre-listing assets contribute nothing and are | |
| excluded from the divisor). | |
| Costs per asset: | |
| - fee 5 bps taker / 2 bps maker on turnover notional | |
| - square-root impact k * sigma_daily * sqrt(Q / ADV) with per-asset ADV | |
| - perp funding drag: position * funding_apr / 365 per day (longs pay; | |
| a constant-APR approximation since funding history does not extend | |
| before 2021 — conservative and honest for a daily backtest) | |
| No-lookahead: strategies pass positions decided at t using data through | |
| t-1 (shift(1) inside the strategy); the engine adds the assert that the | |
| first position day's PnL uses the NEXT bar's return, never the same bar's. | |
| """ | |
| from __future__ import annotations | |
| import numpy as np | |
| import pandas as pd | |
| TAKER_FEE = 0.0005 | |
| MAKER_FEE = 0.0002 | |
| SLIP_FLOOR_BPS = 2.0 | |
| IMPACT_K = 0.5 | |
| FUNDING_APR = 0.1095 # 0.01%/8h baseline, annualized | |
| def realized_daily_vol_bps(close: pd.Series) -> float: | |
| rets = close.pct_change().dropna() | |
| return float(rets.std(ddof=1) * 1e4) | |
| def run_asset_backtest( | |
| close: pd.Series, | |
| positions: pd.Series, | |
| adv_usd: float, | |
| sub_capital: float, | |
| fill_side: str = "taker", | |
| ) -> pd.Series: | |
| """Net daily return series (fraction of sub-capital) for one asset.""" | |
| if not positions.index.equals(close.index): | |
| raise ValueError("positions index must equal close index") | |
| if positions.isna().any(): | |
| raise ValueError("positions contain NaN") | |
| ret = close.pct_change() | |
| # position decided at t earns the t+1 close-to-close return | |
| gross_ret = (positions * ret.shift(-1)).fillna(0.0) | |
| dpos = positions.diff().fillna(positions.iloc[0]) | |
| turnover_frac = dpos.abs() | |
| notional = turnover_frac * sub_capital | |
| fee_rate = TAKER_FEE if fill_side == "taker" else MAKER_FEE | |
| sigma_bps = realized_daily_vol_bps(close) | |
| bps = np.maximum( | |
| SLIP_FLOOR_BPS, IMPACT_K * sigma_bps * np.sqrt(notional / adv_usd) | |
| ) | |
| cost_ret = (notional * (fee_rate + bps / 1e4)) / sub_capital | |
| fund_ret = positions * FUNDING_APR / 365.0 | |
| return gross_ret - cost_ret - fund_ret | |
| def run_panel_backtest( | |
| panel: pd.DataFrame, | |
| positions: pd.DataFrame, | |
| adv: dict[str, float], | |
| initial_capital: float = 1_000_000.0, | |
| fill_side: str = "taker", | |
| ) -> dict: | |
| if not positions.index.equals(panel.index): | |
| raise ValueError("positions index must equal panel index") | |
| if positions.isna().any().any(): | |
| raise ValueError("positions contain NaN") | |
| n_assets = panel.shape[1] | |
| sub_capital = initial_capital / n_assets | |
| per_asset = {} | |
| for sym in panel.columns: | |
| per_asset[sym] = run_asset_backtest( | |
| panel[sym].dropna(), positions[sym].reindex(panel[sym].dropna().index).fillna(0.0), | |
| adv[sym], sub_capital, fill_side, | |
| ) | |
| rets = pd.DataFrame(per_asset) | |
| # capital for not-yet-listed assets sits in CASH (0 return); the | |
| # divisor is ALWAYS the full asset count — dividing by the number of | |
| # listed assets would silently apply leverage in the early era. | |
| portfolio_ret = rets.fillna(0.0).sum(axis=1) / n_assets | |
| return { | |
| "portfolio_return": portfolio_ret, | |
| "per_asset_returns": rets, | |
| } | |