Gahow Wang
0e378efa67
Event-driven long-only simulator with next-open fills, CN tradability (halts, limit-up/down, stamp duty), PIT S&P 500 / CSI 300 membership, rebalance-phase averaging, factor IC scans, maximin stepwise composite search across 5-year blocks, walk-forward LightGBM/ridge combiners, overlays, and deflated Sharpe. Fetchers pull 20y US OHLCV from Yahoo and CSI 300/500 ever-members (membership snapshot from baostock, bars from Yahoo). Research data lives in data/research/ (ignored). Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
56 lines
2.2 KiB
Python
56 lines
2.2 KiB
Python
"""
|
|
Exposure overlays: series in [0, 1] read on signal dates by engine.simulate
|
|
(``exposure=``). The un-invested part earns the panel's cash return.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from research.lab.engine import simulate
|
|
|
|
|
|
def ew_index(p) -> pd.Series:
|
|
"""Equal-weight index of PIT members (daily rebalanced)."""
|
|
r = p.close.pct_change(fill_method=None).where(p.member.shift(1, fill_value=False))
|
|
return (1 + r.mean(axis=1).fillna(0)).cumprod()
|
|
|
|
|
|
def trend(level: pd.Series, ma: int = 200, floor: float = 0.0) -> pd.Series:
|
|
"""1 when the index is above its moving average, else ``floor``."""
|
|
above = level > level.rolling(ma, min_periods=ma).mean()
|
|
return above.astype(float).where(above, floor)
|
|
|
|
|
|
def breadth(p, ma: int = 200, lo: float = 0.4, hi: float = 0.6) -> pd.Series:
|
|
"""Share of members above their own MA mapped linearly from lo→0 to hi→1."""
|
|
c = p.close.ffill()
|
|
above = (c > c.rolling(ma, min_periods=ma).mean()).where(p.member)
|
|
share = above.mean(axis=1)
|
|
return ((share - lo) / (hi - lo)).clip(0, 1)
|
|
|
|
|
|
def drawdown_brake(level: pd.Series, dd: float = 0.15, ma: int = 50) -> pd.Series:
|
|
"""Cut to zero when the index is > dd below its 1y high AND below its MA."""
|
|
high = level.rolling(252, min_periods=60).max()
|
|
bad = (level / high - 1 < -dd) & (level < level.rolling(ma, min_periods=ma).mean())
|
|
return (~bad).astype(float)
|
|
|
|
|
|
def vol_target(p, score, target: float, lookback: int = 60, cap: float = 1.0,
|
|
base_kwargs: dict | None = None, exposure: pd.Series | None = None) -> pd.Series:
|
|
"""Scale exposure by target / realized vol of the unscaled strategy.
|
|
|
|
Two passes: run the strategy fully invested, measure its trailing
|
|
daily-return vol, then size the next rebalance; any given ``exposure``
|
|
(e.g. a trend filter) multiplies the result.
|
|
"""
|
|
kw = dict(base_kwargs or {})
|
|
base = simulate(p, score, **kw)
|
|
rv = base.returns.rolling(lookback, min_periods=lookback // 2).std() * np.sqrt(252)
|
|
scale = (target / rv).clip(upper=cap).fillna(cap)
|
|
if exposure is not None:
|
|
scale = scale * exposure.reindex(scale.index).ffill().fillna(1.0)
|
|
return scale
|