sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 4 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(factor): fast_ops 双操作数join改按位拼接——ts_corr/cov每次调用对10M行帧hash join分钟级(py-spy第五查),同父df全算子保长保序前提下按位等价零成本 [vps]

Browse Source
This commit is contained in:
Claude Dev
2026-08-26 01:57:05 +08:00
parent d995962425
commit 38df909b92
1 changed files with 28 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
sanguo_factor/fast_ops.py
+28 -10
View File
@@ -61,7 +61,12 @@ def fast_ts_corr(feature1: DataProxy, feature2: DataProxy, window: int) -> DataP
Uses correlation identity: corr(x,y) = cov(x,y) / (std_x * std_y)
"""
df_merged = feature1.df.join(feature2.df, on=["datetime", "vt_symbol"])
# 两操作数源自同一父df且全算子保长保序(rolling/shift/cs_rank 均保长)——
# 按位拼接等价于按键join,免 10M 行 hash join(py-spy 实锤分钟级)
h1, h2 = feature1.df.height, feature2.df.height
if h1 != h2:
raise ValueError(f"ts_corr 操作数长度不一致: {h1} vs {h2}(行序对齐前提被破坏)")
df_merged = feature1.df.with_columns(feature2.df["data"].alias("data_right"))
# Use native polars rolling correlation for exact match
corr_result = pl.rolling_corr(
@@ -92,7 +97,11 @@ def fast_ts_cov(feature1: DataProxy, feature2: DataProxy, window: int) -> DataPr
corr_result = fast_ts_corr(feature1, feature2, window)
# Extract std deviations from the merged dataframe
df_merged = feature1.df.join(feature2.df, on=["datetime", "vt_symbol"])
# 两操作数源自同一父df且全算子保长保序——按位拼接等价于按键join
h1, h2 = feature1.df.height, feature2.df.height
if h1 != h2:
raise ValueError(f"ts_cov 操作数长度不一致: {h1} vs {h2}(行序对齐前提被破坏)")
df_merged = feature1.df.with_columns(feature2.df["data"].alias("data_right"))
std_x = pl.col("data").rolling_std(window, min_samples=1, ddof=0).over("vt_symbol")
std_y = pl.col("data_right").rolling_std(window, min_samples=1, ddof=0).over("vt_symbol")
@@ -417,7 +426,12 @@ def fast_ts_corr_v2(feature1: DataProxy, feature2: DataProxy, window: int) -> Da
Uses: corr = (E[xy] - E[x]E[y]) / (std_x * std_y)
All statistics computed via shift-based mean/std
"""
df_merged = feature1.df.join(feature2.df, on=["datetime", "vt_symbol"])
# 两操作数源自同一父df且全算子保长保序(rolling/shift/cs_rank 均保长)——
# 按位拼接等价于按键join,免 10M 行 hash join(py-spy 实锤分钟级)
h1, h2 = feature1.df.height, feature2.df.height
if h1 != h2:
raise ValueError(f"ts_corr_v2 操作数长度不一致: {h1} vs {h2}(行序对齐前提被破坏)")
df_merged = feature1.df.with_columns(feature2.df["data"].alias("data_right"))
x = pl.col("data")
y = pl.col("data_right")
@@ -476,13 +490,17 @@ def fast_ts_cov_v2(feature1: DataProxy, feature2: DataProxy, window: int) -> Dat
std_y_result = fast_ts_std(feature2, window)
# Merge and compute covariance
df_merged = corr_result.df.join(
std_x_result.df.select(["datetime", "vt_symbol", pl.col("data").alias("std_x")]),
on=["datetime", "vt_symbol"]
).join(
std_y_result.df.select(["datetime", "vt_symbol", pl.col("data").alias("std_y")]),
on=["datetime", "vt_symbol"]
)
# 三操作数源自同一父df且全算子保长保序——按位拼接等价于按键join
h_corr, h_x = corr_result.df.height, std_x_result.df.height
if h_corr != h_x:
raise ValueError(f"ts_cov_v2 corr与std_x长度不一致: {h_corr} vs {h_x}")
h_y = std_y_result.df.height
if h_corr != h_y:
raise ValueError(f"ts_cov_v2 corr与std_y长度不一致: {h_corr} vs {h_y}")
df_merged = corr_result.df.with_columns([
std_x_result.df["data"].alias("std_x"),
std_y_result.df["data"].alias("std_y")
])
df = df_merged.select(
pl.col("datetime"),