1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118
| import numpy as np import pandas as pd
prices = pd.DataFrame(columns=["date", "asset", "close", "adj_factor", "volume", "amount"]) fundamentals = pd.DataFrame(columns=["report_period", "announce_date", "asset", "ep", "roe", "asset_growth"]) universe = pd.DataFrame(columns=["date", "asset", "eligible", "industry", "market_cap"])
assets = list("ABCDEF") dates = pd.date_range("2026-01-31", periods=5, freq="ME") ep_values = np.array([0.03, 0.04, 0.05, 0.06, 0.07, 0.08])
fundamentals = pd.DataFrame({ "report_period": pd.to_datetime(["2025-12-31"] * 6), "announce_date": pd.to_datetime(["2026-01-15"] * 6), "asset": assets, "ep": ep_values, "roe": [0.08, 0.10, 0.12, 0.09, 0.15, 0.18], "asset_growth": [0.10, 0.08, 0.06, 0.12, 0.04, 0.02], })
universe_rows = [] for date in dates[:-1]: for i, asset in enumerate(assets): universe_rows.append({ "date": date, "asset": asset, "eligible": True, "industry": "X" if i < 3 else "Y", "market_cap": (100 + 20 * i) * 1_000_000, }) universe = pd.DataFrame(universe_rows, columns=universe.columns)
z_ep = (ep_values - ep_values.mean()) / ep_values.std(ddof=0) close = np.repeat(100.0, 6) price_rows = [] for t, date in enumerate(dates): for i, asset in enumerate(assets): price_rows.append({ "date": date, "asset": asset, "close": close[i], "adj_factor": 1.0, "volume": 1_000_000 + 10_000 * i, "amount": close[i] * (1_000_000 + 10_000 * i), }) if t < len(dates) - 1: next_return = 0.002 + 0.006 * z_ep + 0.001 * np.sin(t + np.arange(6)) close = close * (1 + next_return) prices = pd.DataFrame(price_rows, columns=prices.columns)
assert not prices.duplicated(["date", "asset"]).any() assert not fundamentals.duplicated(["report_period", "announce_date", "asset"]).any() assert not universe.duplicated(["date", "asset"]).any()
prices["adjusted"] = prices["close"] * prices["adj_factor"] prices = prices.sort_values(["asset", "date"]) prices["next_return"] = prices.groupby("asset")["adjusted"].pct_change() prices["signal_date"] = prices.groupby("asset")["date"].shift(1) returns = prices.dropna(subset=["signal_date"]).rename(columns={"date": "return_date"})
panel = universe.rename(columns={"date": "signal_date"}) panel = panel.loc[panel["eligible"]].merge( returns[["signal_date", "return_date", "asset", "next_return"]], on=["signal_date", "asset"], validate="one_to_one", ) panel = panel.merge(fundamentals, on="asset", validate="many_to_one") panel = panel.loc[panel["announce_date"] <= panel["signal_date"]].copy()
assert (panel["announce_date"] <= panel["signal_date"]).all() assert (panel["return_date"] > panel["signal_date"]).all()
panel["group"] = panel.groupby("signal_date")["ep"].transform( lambda x: pd.qcut(x.rank(method="first"), 3, labels=False) ) ic = panel.groupby("signal_date").apply( lambda x: x["ep"].corr(x["next_return"], method="spearman"), include_groups=False, ) group_return = panel.groupby( ["signal_date", "group"], observed=True )["next_return"].mean().unstack() gross_hl = group_return[2] - group_return[0]
cost = pd.Series(0.0005, index=gross_hl.index) net_hl = gross_hl - cost assert (net_hl <= gross_hl + 1e-12).all()
split = len(net_hl) // 2 subsamples = { "first_half": float(net_hl.iloc[:split].mean()), "second_half": float(net_hl.iloc[split:].mean()), } mean_ic = float(ic.mean()) mean_net = float(net_hl.mean()) if mean_ic > 0 and mean_net > 0 and min(subsamples.values()) > 0: conclusion = "因子支持" elif mean_ic <= 0 and mean_net <= 0: conclusion = "因子证伪" else: conclusion = "证据不足"
report = { "question": "点时 EP 是否正向预测下一月净收益", "rows": len(panel), "coverage": len(panel) / len(universe), "mean_rank_ic": mean_ic, "mean_gross_hl": float(gross_hl.mean()), "mean_net_hl": mean_net, "subsamples": subsamples, "failure_conditions": [ "公告时点不合格", "任一子样本净收益非正", "成本或微盘排除后结论消失", ], "conclusion": conclusion, } print(report)
|