claude_dev
404330f29a
旧实现 λI 直接加在被 WLS 权重放大的原始正规阵上: 真实 √市值 量级 (w≈6e4, diag(A)~1e8)下相对收缩 ~1e-9, ridge 解≡OLS 解, 对照通道在 唯一触发场景(VIF>10)下必然 ridge_flip 恒空静默放行——spec §4.7-3 「标准化设计阵固定档」名不副实。 修法: 惩罚随各列方差缩放 A+λ·diag(A)(数学上等价 D^½(R+λI)D^½ 的 标准化解)——权重整体缩放 c 时解不变(尺度不变性), 等权/√市值双通道 同效; 系数统一收缩 1/(1+λ), 截距不罚。当前 quant12 vif≈1.4 不进 触发区, 已产归因件不受影响(通道仅 VIF>10 才跑)。 测试 427→428: ①尺度不变+真实量级生效双性质钉死(旧 no-op 必红); ②flip 测试从 isinstance-only 强化为多路共线构造下非空断言 (src4=三源等权组合: 两两 ρ≈0.58 不并族但对前三源 R²≈1, VIF 全线 34~143; OLS 把 premia 为正的 src1 窗累计拧负, ridge 翻正捕获)。 因子档 §7.5 ridge 语义同步修订(档随码走)。 Co-Authored-By: Claude Code <noreply@anthropic.com>
433 lines
21 KiB
Python
433 lines
21 KiB
Python
# tests/factor/test_attribution.py
|
|
"""因子贡献分解 TDD(P3 第三刀, spec §4.7)."""
|
|
import os
|
|
import sys
|
|
|
|
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
|
|
def test_size_log_registered():
|
|
"""控制集数据件: size_log = log(close*share_capital) 可评估可导出."""
|
|
from sanguo_factor import fundamental_library
|
|
from sanguo_factor.registry import get_factor
|
|
fundamental_library._register_all() # 幂等重挂(先序测试 clear 后可重建)
|
|
fac = get_factor("size_log")
|
|
assert fac is not None
|
|
assert fac["expression"] == "log(close * share_capital)"
|
|
|
|
|
|
# ---------------- Task 2: 装载层 ----------------
|
|
|
|
def test_load_sections_and_forward_returns(tmp_path):
|
|
from sanguo_factor import attribution as att
|
|
dates = pd.bdate_range("2025-01-02", periods=5)
|
|
df = pd.DataFrame({"600000.SSE": [1.0, 2.0, 4.0, 8.0, 16.0]}, index=dates)
|
|
df.to_parquet(tmp_path / "src1.parquet")
|
|
got = att.load_sections(str(tmp_path), ["src1"])
|
|
assert list(got) == ["src1"] and got["src1"].shape == (5, 1)
|
|
with pytest.raises(FileNotFoundError):
|
|
att.load_sections(str(tmp_path), ["nope"])
|
|
r = att.forward_returns(df)
|
|
assert np.isclose(r["600000.SSE"].iloc[-1], 1.0) # 8→16 = +100%
|
|
|
|
|
|
def test_load_closes_from_sqlite(tmp_path):
|
|
import sqlite3
|
|
from sanguo_factor import attribution as att
|
|
db = str(tmp_path / "qt.db")
|
|
conn = sqlite3.connect(db)
|
|
conn.execute("CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT,"
|
|
" interval TEXT, close_price REAL)")
|
|
for d, c in [("2025-01-02 00:00:00", 10.0), ("2025-01-03 00:00:00", 11.0)]:
|
|
conn.execute("INSERT INTO dbbardata VALUES('600000','SSE',?,'d',?)", (d, c))
|
|
conn.commit(); conn.close()
|
|
got = att.load_closes(db, ["600000.SSE"], "2025-01-01", "2025-01-31")
|
|
assert list(got.index) == ["2025-01-02", "2025-01-03"]
|
|
assert got["600000.SSE"].tolist() == [10.0, 11.0]
|
|
|
|
|
|
# ---------------- Task 3: 族代表 + VIF ----------------
|
|
|
|
def _expo(n_days=120, n_stocks=40, seed=3):
|
|
rng = np.random.default_rng(seed)
|
|
syms = [f"{600000 + i}.SSE" for i in range(n_stocks)]
|
|
dates = pd.bdate_range("2025-01-02", periods=n_days)
|
|
return pd.DataFrame(rng.normal(0, 1, (n_days, n_stocks)),
|
|
index=dates, columns=syms)
|
|
|
|
|
|
def test_family_representatives_clusters_and_medoid():
|
|
from sanguo_factor import attribution as att
|
|
base = _expo()
|
|
copy_ = base + _expo(seed=9) * 0.05 # 与 base 相关≈0.99
|
|
far = _expo(seed=11) # 独立
|
|
fams = att.family_representatives({"src_a": base, "src_a2": copy_, "src_b": far})
|
|
reps = sorted(fams)
|
|
assert len(reps) == 2 # 压成两族
|
|
members = sorted(sum(fams.values(), []))
|
|
assert members == ["src_a", "src_a2", "src_b"]
|
|
# 代表=中位点: src_a 与 src_a2 平手按名序 → src_a
|
|
assert "src_a" in reps and fams["src_a"] == ["src_a", "src_a2"]
|
|
|
|
|
|
def test_vif_flags_collinearity():
|
|
from sanguo_factor import attribution as att
|
|
base = _expo()
|
|
# a_copy 与 a 相关≈0.995(真噪声; +常数会使 float64 相关精确=1 → 奇异
|
|
# → inf, 那是 vif 的正确语义但测不了数值分支)
|
|
design = pd.DataFrame({"a": base.stack().to_numpy(),
|
|
"a_copy": (base + _expo(seed=9) * 0.1).stack().to_numpy(),
|
|
"c": _expo(seed=11).stack().to_numpy()})
|
|
got = att.vif(design)
|
|
assert got["a_copy"] > 10 and got["c"] < 5
|
|
|
|
|
|
# ---------------- Task 4: 单日 Fama-MacBeth WLS + ridge 对照 ----------------
|
|
|
|
def test_solve_day_recovers_planted_premia():
|
|
"""合成世界: y = 1*0.5 + x1*f1 + x2*f2 + e → WLS 回收 f1/f2."""
|
|
from sanguo_factor import attribution as att
|
|
rng = np.random.default_rng(5)
|
|
n = 2000
|
|
x1, x2 = rng.normal(0, 1, n), rng.normal(0, 1, n)
|
|
f1, f2 = 0.004, -0.002
|
|
y = 0.0005 + x1 * f1 + x2 * f2 + rng.normal(0, 0.002, n)
|
|
X = np.column_stack([np.ones(n), x1, x2])
|
|
w = rng.uniform(0.5, 2.0, n) # sqrt(市值) 形状的任意权重
|
|
beta = att.solve_day(X, y, w)
|
|
assert abs(beta[0] - 0.0005) < 0.0004
|
|
assert abs(beta[1] - f1) < 0.0004 and abs(beta[2] - f2) < 0.0004
|
|
# ridge 不罚截距(仍贴真值; 斜率收缩后截距吸收残余, 不逐位冻结);
|
|
# 系数统一收缩 1/(1+λ)(相关形标准化语义, 09-27 审计 F-2 修后——
|
|
# 旧「对强信号收缩很小」实为权重放大下的 no-op)
|
|
br = att.solve_day(X, y, w, ridge=att.RIDGE_LAMBDA)
|
|
assert abs(br[0] - 0.0005) < 0.0005
|
|
assert abs(br[1] - f1 / (1 + att.RIDGE_LAMBDA)) < 0.0004
|
|
|
|
|
|
def test_solve_day_ridge_scale_invariant_and_effective():
|
|
"""审计 F-2 回归: 旧实现 λI 加在被 √市值 权重放大的原始正规阵上
|
|
(对角 ~1e8, 相对收缩 ~1e-9), 真实量级下对照通道恒等价 OLS=no-op,
|
|
ridge_flip 必然恒空. 修后惩罚随各列方差缩放, 两个性质钉死:
|
|
①权重整体缩放不变(尺度不变性=修复定义);
|
|
②真实 √市值 量级(e^22 元)下对照实质生效(相对差 >> 旧 no-op)."""
|
|
from sanguo_factor import attribution as att
|
|
rng = np.random.default_rng(11)
|
|
n = 3000
|
|
z1 = rng.standard_normal(n)
|
|
z2 = 0.98 * z1 + np.sqrt(1 - 0.98 ** 2) * rng.standard_normal(n) # ρ≈0.98
|
|
X = np.column_stack([np.ones(n), z1, z2])
|
|
beta_true = np.array([0.0005, 0.002, -0.001])
|
|
y = X @ beta_true + rng.standard_normal(n) * 0.01
|
|
size_log = rng.normal(22.0, 0.8, n) # 真实量级: 市值=e^22≈3.6e9 元
|
|
w = np.sqrt(np.exp(size_log)) # 调用链同款: w=√市值
|
|
b0 = att.solve_day(X, y, w, ridge=0.0)
|
|
br = att.solve_day(X, y, w, ridge=att.RIDGE_LAMBDA)
|
|
br2 = att.solve_day(X, y, w * 7.3, ridge=att.RIDGE_LAMBDA)
|
|
assert np.allclose(br, br2, rtol=1e-9), "权重整体缩放不得改变 ridge 解"
|
|
rel = float(np.abs(b0 - br).max() / np.abs(b0).max())
|
|
assert rel > 0.05, f"ridge 相对差 {rel:.2e} 仍是 no-op(旧病灶回潮)"
|
|
|
|
|
|
def test_solve_day_weights_actually_weight():
|
|
"""WLS 真加权: 大权重点主导拟合(OLS 会拖走)."""
|
|
from sanguo_factor import attribution as att
|
|
x = np.array([0.0, 1.0, 2.0, 3.0])
|
|
y = np.array([0.0, 0.006, 0.02, 0.036]) # 斜率≈0.012
|
|
X = np.column_stack([np.ones(4), x])
|
|
w_hi = np.array([50.0, 50.0, 0.01, 0.01]) # 押前两点(斜率≈0.006)
|
|
beta = att.solve_day(X, y, w_hi)
|
|
assert abs(beta[1] - 0.006) < 0.001
|
|
|
|
|
|
# ---------------- Task 5: 持仓重建 + 组合收益(漂移口径) ----------------
|
|
|
|
def test_holdings_schedule_grid_topn_and_kcbj():
|
|
from sanguo_factor import attribution as att
|
|
dates = pd.bdate_range("2025-01-02", periods=10)
|
|
sec = pd.DataFrame(index=dates, columns=["600000.SSE", "000001.SZSE",
|
|
"688001.SSE"], dtype=float)
|
|
sec["600000.SSE"] = 2.0
|
|
sec["000001.SZSE"] = 1.0
|
|
sec["688001.SSE"] = 9.0 # 科创: 值最大但须被剔
|
|
sched = att.holdings_schedule(sec, top_n=1, reb_days=5)
|
|
assert len(sched) == 10
|
|
d0, h0 = sched[0]
|
|
assert h0 == ["600000.SSE"] # 688 剔除后取最大
|
|
assert sched[4][1] == ["600000.SSE"] # 第 5 日仍在窗内
|
|
assert sched[5][1] == ["600000.SSE"] # 第 6 日=重选(同一只)
|
|
|
|
|
|
def test_portfolio_returns_drift_and_rebalance():
|
|
from sanguo_factor import attribution as att
|
|
dates = pd.bdate_range("2025-01-02", periods=3)
|
|
rets = pd.DataFrame({"a": [np.nan, 0.10, 0.0],
|
|
"b": [np.nan, -0.10, 0.0]}, index=dates)
|
|
sched = [(dates[1], ["a", "b"]), (dates[2], ["a", "b"])]
|
|
r_p, w = att.portfolio_returns(sched, rets)
|
|
assert np.isclose(r_p.loc[dates[1]], 0.0) # 等权 ±10% 对冲
|
|
# 漂移: a 涨 b 跌 → 次日 a 权重 1.1/2.0
|
|
assert np.isclose(w.loc[dates[2], "a"], 0.55)
|
|
assert np.isclose(r_p.loc[dates[2]], 0.0)
|
|
|
|
|
|
# ---------------- Task 6: 归因主循环(恒等式+size 锚点+ridge 对照) ----------------
|
|
|
|
def _planted_world(n_days=260, n_stocks=60, seed=13, rho=0.99):
|
|
"""合成两因子世界: r_t = mkt_t + x1_{t-1}·f1_t + x2_{t-1}·f2_t + e_t.
|
|
|
|
暴露须 AR(1) 持续(rho=0.99, 21 日持有期相关≈0.81)——逐日独立噪声的
|
|
暴露会让持有期组合对当日暴露≈0(真实 quant12 暴露的持续性正是 63 日
|
|
调仓可行的原因), 回收测试就归不到因子上(v1 教训)."""
|
|
rng = np.random.default_rng(seed)
|
|
syms = [f"{600000 + i}.SSE" for i in range(n_stocks)]
|
|
dates = pd.bdate_range("2025-01-02", periods=n_days)
|
|
|
|
def persistent():
|
|
x = rng.normal(0, 1, (n_days, n_stocks))
|
|
for t in range(1, n_days):
|
|
x[t] = rho * x[t - 1] + np.sqrt(1 - rho * rho) * x[t]
|
|
return pd.DataFrame(x, index=dates, columns=syms)
|
|
|
|
f1 = rng.normal(0.0010, 0.004, n_days)
|
|
f2 = rng.normal(-0.0005, 0.003, n_days)
|
|
mkt = rng.normal(0.0003, 0.008, n_days)
|
|
x1, x2 = persistent(), persistent()
|
|
e = pd.DataFrame(rng.normal(0, 0.01, (n_days, n_stocks)), index=dates, columns=syms)
|
|
lag1, lag2 = x1.shift(1).fillna(0.0), x2.shift(1).fillna(0.0)
|
|
r = pd.DataFrame(mkt[:, None] + lag1.to_numpy() * f1[:, None]
|
|
+ lag2.to_numpy() * f2[:, None] + e.to_numpy(),
|
|
index=dates, columns=syms)
|
|
closes = (1.0 + r).cumprod()
|
|
return dict(syms=syms, dates=dates, x1=x1, x2=x2, r=r, closes=closes,
|
|
f1=f1, f2=f2, mkt=mkt)
|
|
|
|
|
|
def _industry_map(syms):
|
|
return pd.DataFrame({"symbol": syms,
|
|
"industry_code": [1 + i % 5 for i in range(len(syms))]})
|
|
|
|
|
|
def test_attribute_identity_and_recovery():
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world()
|
|
rets = att.forward_returns(w["closes"])
|
|
size = pd.DataFrame(np.random.default_rng(2).normal(22.0, 0.8,
|
|
(len(w["dates"]), len(w["syms"]))),
|
|
index=w["dates"], columns=w["syms"])
|
|
score = att.zscore_cross(w["x1"]) + att.zscore_cross(w["x2"])
|
|
res = att.attribute(score, {"src1": w["x1"], "src2": w["x2"]},
|
|
size, _industry_map(w["syms"]), rets,
|
|
top_n=20, reb_days=21)
|
|
# ① 恒等式: Σ贡献(含 market/size/industry/specific) ≡ 组合收益
|
|
total = res["contributions"].sum(axis=1)
|
|
assert np.allclose(total, res["portfolio_ret"], atol=1e-10)
|
|
# ② 回收: 两源累计贡献与植入 premia 同号且量级对
|
|
c1 = res["contributions"]["src1"].sum()
|
|
c2 = res["contributions"]["src2"].sum()
|
|
assert c1 > 0.02 and c2 < -0.01
|
|
# ③ size 无植入效应 → 泄漏锚点不报警
|
|
leak, cum = att.size_leak(res)
|
|
assert leak is False and abs(cum) < 0.02
|
|
# ④ 诊断齐 + contributions 无 ind_ 明细列(infra 修正①: schema 不泄漏)
|
|
assert 0.0 < res["r2_mean"] <= 1.0 and "src1" in res["vif"]
|
|
assert not any(c.startswith("ind_") for c in res["contributions"].columns)
|
|
assert res["families"]
|
|
|
|
|
|
def test_attribute_nan_return_holding_keeps_identity():
|
|
"""infra 修正②: 停牌股 NaN 收益不得穿透进 specific(价格冻结=0)."""
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world(n_days=120)
|
|
rets = att.forward_returns(w["closes"])
|
|
held_sym = w["syms"][0]
|
|
rets[held_sym] = rets[held_sym].where(
|
|
np.arange(len(rets)) % 17 != 3) # 散布 NaN(停牌日)
|
|
# size 须截面有方差(常数 → zscore 全 NaN → 主循环空转, 断言空转也过)
|
|
size = pd.DataFrame(np.random.default_rng(8).normal(22.0, 0.8,
|
|
(len(w["dates"]), len(w["syms"]))),
|
|
index=w["dates"], columns=w["syms"])
|
|
score = att.zscore_cross(w["x1"]) + att.zscore_cross(w["x2"])
|
|
res = att.attribute(score, {"src1": w["x1"], "src2": w["x2"]},
|
|
size, _industry_map(w["syms"]), rets,
|
|
top_n=20, reb_days=21)
|
|
total = res["contributions"].sum(axis=1)
|
|
assert not total.isna().any() # NaN 穿透即炸点
|
|
assert np.allclose(total, res["portfolio_ret"], atol=1e-10)
|
|
|
|
|
|
def test_industry_join_miss_bucket_zero():
|
|
"""data 行业表契约: 表未覆盖股 join 不中 → 0 码独立桶, 不静默丢样本.
|
|
|
|
覆盖 55/60(未映射 5 只 < 各行业桶) → ind_0=显式哑变量列; 回归样本
|
|
含未映射股, unknown_industry_share 诊断可数; 恒等式不缺块.
|
|
"""
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world(n_days=120)
|
|
rets = att.forward_returns(w["closes"])
|
|
# size 须截面有方差: 全常数 → zscore 除 std=0 全 NaN, 主循环空转
|
|
size = pd.DataFrame(np.random.default_rng(7).normal(22.0, 0.8,
|
|
(len(w["dates"]), len(w["syms"]))),
|
|
index=w["dates"], columns=w["syms"])
|
|
score = att.zscore_cross(w["x1"]) + att.zscore_cross(w["x2"])
|
|
covered = w["syms"][:-5] # 55 只已映射, 5 只 join 不中
|
|
ind = pd.DataFrame({"symbol": covered,
|
|
"industry_code": [1 + i % 5 for i in range(len(covered))]})
|
|
res = att.attribute(score, {"src1": w["x1"], "src2": w["x2"]},
|
|
size, ind, rets, top_n=20, reb_days=21)
|
|
assert "ind_0" in res["factor_returns"].columns # 0 码=显式哑变量列
|
|
assert res["unknown_industry_share"] > 0.0 # 未映射计入诊断
|
|
total = res["contributions"].sum(axis=1)
|
|
assert np.allclose(total, res["portfolio_ret"], atol=1e-10)
|
|
|
|
|
|
def test_zscore_cross_inf_row():
|
|
"""真数据首跑实证: alpha83 部分日行内 ±inf → std=NaN 毁整行 z.
|
|
口径: inf 暴露=缺失(NaN), 其余股正常 z 化."""
|
|
from sanguo_factor import attribution as att
|
|
df = pd.DataFrame({"a": [1.0, np.inf], "b": [2.0, 3.0], "c": [3.0, -np.inf],
|
|
"d": [np.inf, 7.0], "e": [5.0, 11.0]})
|
|
z = att.zscore_cross(df)
|
|
assert np.isfinite(z.loc[0, "b"]) # 无 inf 行完全正常
|
|
assert np.isnan(z.loc[0, "d"]) # 行 0 inf 位=d 列
|
|
assert np.isfinite(z.loc[1, "b"]) # inf 行其余股不再陪葬
|
|
assert np.isnan(z.loc[1, "a"]) and np.isnan(z.loc[1, "c"])
|
|
|
|
|
|
def test_industry_symbol_suffix_normalized():
|
|
"""data 表交付 .SH/.SZ 通用形态(SW 官方, 非 vnpy vt 形): 消费端
|
|
归一后 join 命中——不归一则全体落 0 码, 行业控制失效."""
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world(n_days=120)
|
|
rets = att.forward_returns(w["closes"])
|
|
size = pd.DataFrame(np.random.default_rng(10).normal(22.0, 0.8,
|
|
(len(w["dates"]), len(w["syms"]))),
|
|
index=w["dates"], columns=w["syms"])
|
|
score = att.zscore_cross(w["x1"]) + att.zscore_cross(w["x2"])
|
|
ind = pd.DataFrame({
|
|
"symbol": [s.replace(".SSE", ".SH") for s in w["syms"]],
|
|
"industry_code": [1 + i % 5 for i in range(len(w["syms"]))],
|
|
})
|
|
res = att.attribute(score, {"src1": w["x1"], "src2": w["x2"]},
|
|
size, ind, rets, top_n=20, reb_days=21)
|
|
assert res["unknown_industry_share"] == 0.0 # 全命中, 无 0 码
|
|
assert "ind_0" not in res["factor_returns"].columns
|
|
total = res["contributions"].sum(axis=1)
|
|
assert np.allclose(total, res["portfolio_ret"], atol=1e-10)
|
|
|
|
|
|
def test_attribute_ridge_flip_detected():
|
|
"""多路共线(VIF>10) → ridge 对照真实翻转(非空)——审计 F-2 补强.
|
|
|
|
构造: src1/src2/src3 独立, src4=三源等权组合——两两 ρ≈0.58<0.7 不并族,
|
|
但 src4 对前三源 R²≈1 → VIF 全线 34~143 超阈. OLS 在病态共线下把
|
|
premia 为正的 src1 窗累计拧成负值, ridge 收缩后翻正——通道捕获.
|
|
旧 no-op 惩罚(λI 加在权重放大的原始阵上)下 ridge≡OLS, 本测试必红.
|
|
"""
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world(seed=21)
|
|
rets = att.forward_returns(w["closes"])
|
|
# size 须截面有方差(常数 → zscore 全 NaN → 主循环空转);
|
|
# N(22,0.8)=真实 log 市值量级(e^22≈3.6e9 元), 尺度不变性在此量级下验证
|
|
size = pd.DataFrame(np.random.default_rng(9).normal(22.0, 0.8,
|
|
(len(w["dates"]), len(w["syms"]))),
|
|
index=w["dates"], columns=w["syms"])
|
|
rng = np.random.default_rng(21)
|
|
s3 = pd.DataFrame(rng.normal(0, 1, w["x1"].shape),
|
|
index=w["dates"], columns=w["syms"])
|
|
noise3 = pd.DataFrame(rng.normal(0, 1, w["x1"].shape),
|
|
index=w["dates"], columns=w["syms"])
|
|
s4 = (w["x1"] + w["x2"] + s3) / np.sqrt(3.0) + noise3 * 0.03
|
|
score = att.zscore_cross(w["x1"])
|
|
res = att.attribute(score, {"src1": w["x1"], "src2": w["x2"],
|
|
"src3": s3, "src4": s4},
|
|
size, _industry_map(w["syms"]), rets,
|
|
top_n=20, reb_days=21)
|
|
assert any(v > att.VIF_TRIGGER for v in res["vif"].values())
|
|
assert res["ridge_flip"], "共线已超阈, ridge 对照须捕获翻转(旧 no-op 恒空)"
|
|
assert "src1" in res["ridge_flip"] # OLS 拧负/ridge 翻正实证代表
|
|
|
|
|
|
# ---------------- Task 7: 月度汇总 + 滚动 IR + 报告 JSON + CLI ----------------
|
|
|
|
def test_monthly_rows_schema_verbatim():
|
|
from sanguo_factor import attribution as att
|
|
dates = pd.date_range("2025-01-31", periods=12, freq="ME")
|
|
contrib = pd.DataFrame({"src1": 0.001, "market": 0.0005, "size": 0.0,
|
|
"industry": 0.0, "specific": 0.0002},
|
|
index=dates)
|
|
res = {"contributions": contrib,
|
|
"exposure_avg": {"src1": 0.3, "size": 0.0},
|
|
"r2_mean": 0.5, "vif": {"src1": 2.0}, "ridge_flip": [],
|
|
"nan_exposure_holding_share": 0.0, "unknown_industry_share": 0.0}
|
|
rows = att.monthly_rows(res, families={"src1": ["src1"]})
|
|
assert rows[0] == {"factor": "src1", "month": "2025-01",
|
|
"contribution_ret": 0.001,
|
|
"cum_ret": float((1.001 ** 12) - 1),
|
|
"exposure_avg": 0.3}
|
|
# 每月行序: src1 → market → size → industry → specific(三桶殿后)
|
|
last_month_rows = [r for r in rows if r["month"] == "2025-12"]
|
|
assert [r["factor"] for r in last_month_rows] == \
|
|
["src1", "market", "size", "industry", "specific"]
|
|
assert set(rows[0]) == {"factor", "month", "contribution_ret",
|
|
"cum_ret", "exposure_avg"}
|
|
|
|
|
|
def test_rolling_ir_annualized():
|
|
from sanguo_factor import attribution as att
|
|
rng = np.random.default_rng(6)
|
|
s = pd.Series(rng.normal(0.001, 0.01, 504),
|
|
index=pd.bdate_range("2024-01-02", periods=504))
|
|
ir = att.rolling_ir(s, months=12)
|
|
assert ir is not None and -5.0 < ir < 5.0
|
|
assert att.rolling_ir(s.iloc[:10], months=12) is None
|
|
|
|
|
|
def test_main_end_to_end(tmp_path):
|
|
"""CLI 端到端: 合成 sections+行业 parquet+sqlite closes → JSON 落盘."""
|
|
import json
|
|
import sqlite3
|
|
from sanguo_factor import attribution as att
|
|
w = _planted_world(n_days=120)
|
|
score = att.zscore_cross(w["x1"]) + att.zscore_cross(w["x2"])
|
|
score.to_parquet(tmp_path / "comp.parquet")
|
|
w["x1"].to_parquet(tmp_path / "src1.parquet")
|
|
w["x2"].to_parquet(tmp_path / "src2.parquet")
|
|
size = pd.DataFrame(22.0, index=w["dates"], columns=w["syms"])
|
|
size.to_parquet(tmp_path / "size_log.parquet")
|
|
db = str(tmp_path / "qt.db")
|
|
conn = sqlite3.connect(db)
|
|
conn.execute("CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT,"
|
|
" interval TEXT, close_price REAL)")
|
|
for s in w["syms"]:
|
|
code = s.split(".")[0]
|
|
for d, c in zip(w["dates"], w["closes"][s]):
|
|
conn.execute("INSERT INTO dbbardata VALUES(?, 'SSE', ?, 'd', ?)",
|
|
(code, f"{d.date()} 00:00:00", float(c)))
|
|
conn.commit(); conn.close()
|
|
pd.DataFrame({"symbol": w["syms"],
|
|
"industry_code": [1 + i % 5 for i in range(len(w["syms"]))],
|
|
"industry_name": ["一", "二", "三", "四", "五"] * 12}
|
|
).to_parquet(tmp_path / "ind.parquet")
|
|
out = tmp_path / "out"; out.mkdir()
|
|
rc = att.main(["--sections-dir", str(tmp_path), "--factor", "comp",
|
|
"--sources", "src1,src2",
|
|
"--start", str(w["dates"][0].date()),
|
|
"--end", str(w["dates"][-1].date()),
|
|
"--vnpy-db", db, "--industry-path", str(tmp_path / "ind.parquet"),
|
|
"--out-dir", str(out), "--top-n", "20", "--reb-days", "21"])
|
|
assert rc == 0
|
|
jf = out / f"attribution_{w['dates'][-1].date()}.json"
|
|
assert jf.exists() # 文件名=--end 末日(动态取)
|
|
doc = json.loads(jf.read_text())
|
|
assert doc["factorContribution"] and isinstance(doc["factorContribution"], list)
|
|
assert set(doc["factorContribution"][0]) == \
|
|
{"factor", "month", "contribution_ret", "cum_ret", "exposure_avg"}
|
|
assert doc["diagnostics"]["size_leak"] is False
|
|
assert not any(r["factor"].startswith("ind_")
|
|
for r in doc["factorContribution"])
|