diff --git a/data/universe.py b/data/universe.py index 9116467..a026e9e 100644 --- a/data/universe.py +++ b/data/universe.py @@ -1,4 +1,4 @@ -"""CSI 300 (HS300) universe helpers.""" +"""CSI 300 (HS300) and CSI 500 (ZZ500) universe helpers.""" import logging import baostock as bs @@ -19,6 +19,20 @@ SYMBOLS = [ ] +# First 30 CSI 500 (ZZ500) constituents (mid/small caps) in 'shXXXXXX' / +# 'szXXXXXX' format. Hardcoded for fast, deterministic smoke tests. Use +# get_zz500_stocks() for the live, full list. Mean reversion tends to be +# stronger in these smaller caps than in the HS300 large caps. +CSI500_SYMBOLS = [ + "sh600006", "sh600008", "sh600017", "sh600020", "sh600021", + "sh600026", "sh600037", "sh600039", "sh600053", "sh600056", + "sh600060", "sh600061", "sh600062", "sh600073", "sh600089", + "sh600095", "sh600118", "sh600125", "sh600126", "sh600143", + "sh600153", "sh600160", "sh600169", "sh600176", "sh600183", + "sz000009", "sz000012", "sz000021", "sz000025", "sz000027", +] + + def get_hs300_stocks() -> pd.DataFrame: """Fetch the current CSI 300 constituents from baostock. @@ -36,3 +50,22 @@ def get_hs300_stocks() -> pd.DataFrame: df = pd.DataFrame(stocks, columns=["code", "name", "date"]) df["code"] = df["code"].str.replace(".", "", regex=False) return df + + +def get_zz500_stocks() -> pd.DataFrame: + """Fetch the current CSI 500 (ZZ500) constituents from baostock. + + Returns: + DataFrame with columns ``code`` (e.g. ``sh600006``), ``name``, ``date``. + """ + bs.login() + try: + rs = bs.query_zz500_stocks() + stocks = [] + while rs.next(): + stocks.append(rs.get_row_data()) + finally: + bs.logout() + df = pd.DataFrame(stocks, columns=["code", "name", "date"]) + df["code"] = df["code"].str.replace(".", "", regex=False) + return df diff --git a/run_example.py b/run_example.py index 165dd46..faa210a 100644 --- a/run_example.py +++ b/run_example.py @@ -1,6 +1,12 @@ #!/usr/bin/env python3 -"""End-to-end pipeline: HS300 universe -> reversal signal -> cross-sectional IC --> multi-stock backtest (AlphaStrategy + RankEqualWeightBuilder) -> reports.""" +"""End-to-end pipeline: universe -> signal -> cross-sectional IC +-> multi-stock backtest (AlphaStrategy + RankEqualWeightBuilder) -> reports. + +Usage: + python3 run_example.py --universe hs300 --signal reversal + python3 run_example.py --universe csi500 --signal reversal_vol +""" +import argparse import logging import pandas as pd @@ -9,10 +15,11 @@ from analysis.report import generate_report from backtest.config import BacktestConfig from backtest.runner import BacktestRunner from data.downloader import download_batch -from data.universe import SYMBOLS +from data.universe import SYMBOLS, CSI500_SYMBOLS from eval.metrics import evaluate_cross_sectional from portfolio.builder import RankEqualWeightBuilder from signals.reversal import ReversalSignal +from signals.reversal_vol import ReversalVolSignal from strategies.alpha_strategy import AlphaStrategy logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s") @@ -29,18 +36,27 @@ def _forward_returns(data: dict[str, pd.DataFrame], horizon: int) -> pd.DataFram return pd.DataFrame(forward_returns) -def main(forward_horizon: int = 5): - symbols = SYMBOLS[:30] +def main(forward_horizon: int = 5, universe: str = "csi500", signal_name: str = "reversal_vol"): + universes = {"hs300": SYMBOLS, "csi500": CSI500_SYMBOLS} + symbols = universes.get(universe, CSI500_SYMBOLS)[:30] + + signals = { + "reversal": ReversalSignal(lookback=5), + "reversal_vol": ReversalVolSignal(lookback=5, vol_window=20), + } + signal = signals.get(signal_name, ReversalVolSignal(lookback=5, vol_window=20)) + start, end = "2023-01-01", "2024-12-31" initial_cash = 1_000_000 + logger.info(f"Universe: {universe} ({len(symbols)} stocks), Signal: {signal.name}") + # 1-2. Download daily data for the universe. data = download_batch(symbols, start, end) data = {s: df for s, df in data.items() if df is not None and not df.empty} logger.info(f"Downloaded {len(data)}/{len(symbols)} symbols") - # 3. Compute the reversal signal per stock. - signal = ReversalSignal(lookback=5) + # 3. Compute the signal per stock. signal_series: dict[str, pd.Series] = {} for sym, df in data.items(): sig = signal.compute(df) @@ -52,7 +68,7 @@ def main(forward_horizon: int = 5): returns_df = _forward_returns(data, forward_horizon) signal_eval = evaluate_cross_sectional(signals_df, returns_df) - # 4b. Multi-horizon IC to show which horizon the signal works at. + # 4b. Multi-horizon IC. horizon_evals = { h: evaluate_cross_sectional(signals_df, _forward_returns(data, h)) for h in (1, 5, 20) @@ -84,6 +100,7 @@ def main(forward_horizon: int = 5): # 8. Print summary. print("\nSIGNAL IC") print("=" * 50) + print(f"Universe: {universe} | Signal: {signal.name}") print(f"IC mean / std / IR: {signal_eval['ic_mean']:.4f} / " f"{signal_eval['ic_std']:.4f} / {signal_eval['ir']:.4f}") print(f"Rank IC mean / std / IR: {signal_eval['rank_ic_mean']:.4f} / " @@ -102,4 +119,8 @@ def main(forward_horizon: int = 5): if __name__ == "__main__": - main() + parser = argparse.ArgumentParser(description="Chinese equity quant backtest") + parser.add_argument("--universe", default="csi500", choices=["hs300", "csi500"]) + parser.add_argument("--signal", default="reversal_vol", choices=["reversal", "reversal_vol"]) + args = parser.parse_args() + main(universe=args.universe, signal_name=args.signal) diff --git a/signals/__init__.py b/signals/__init__.py index 4fc8c97..6b2ba11 100644 --- a/signals/__init__.py +++ b/signals/__init__.py @@ -1,5 +1,6 @@ """Alpha signal abstractions.""" from signals.base import AlphaSignal from signals.reversal import ReversalSignal +from signals.reversal_vol import ReversalVolSignal -__all__ = ["AlphaSignal", "ReversalSignal"] +__all__ = ["AlphaSignal", "ReversalSignal", "ReversalVolSignal"] diff --git a/signals/reversal_vol.py b/signals/reversal_vol.py new file mode 100644 index 0000000..f9ea5f0 --- /dev/null +++ b/signals/reversal_vol.py @@ -0,0 +1,27 @@ +"""Volatility-scaled short-horizon reversal signal.""" +import pandas as pd + +from signals.base import AlphaSignal + + +class ReversalVolSignal(AlphaSignal): + """Reversal score normalized by trailing volatility. + + The raw reversal ``-close.pct_change(lookback)`` is divided by the rolling + standard deviation of daily returns over ``vol_window``. Scaling by + volatility damps the score of noisy, high-vol names so the signal favors + oversold stocks whose move is large *relative* to their own volatility. + """ + + def __init__(self, lookback: int = 5, vol_window: int = 20): + self.lookback = lookback + self.vol_window = vol_window + + def compute(self, df: pd.DataFrame) -> pd.Series: + reversal = -df["close"].pct_change(self.lookback) + vol = df["close"].pct_change().rolling(self.vol_window).std() + return reversal / vol + + @property + def name(self) -> str: + return f"reversal_vol_{self.lookback}d_{self.vol_window}d"