From 38df909b920718535802040a08c6b0d39454bde7 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 26 Aug 2026 01:57:05 +0800 Subject: [PATCH] =?UTF-8?q?perf(factor):=20fast=5Fops=20=E5=8F=8C=E6=93=8D?= =?UTF-8?q?=E4=BD=9C=E6=95=B0join=E6=94=B9=E6=8C=89=E4=BD=8D=E6=8B=BC?= =?UTF-8?q?=E6=8E=A5=E2=80=94=E2=80=94ts=5Fcorr/cov=E6=AF=8F=E6=AC=A1?= =?UTF-8?q?=E8=B0=83=E7=94=A8=E5=AF=B910M=E8=A1=8C=E5=B8=A7hash=20join?= =?UTF-8?q?=E5=88=86=E9=92=9F=E7=BA=A7(py-spy=E7=AC=AC=E4=BA=94=E6=9F=A5),?= =?UTF-8?q?=E5=90=8C=E7=88=B6df=E5=85=A8=E7=AE=97=E5=AD=90=E4=BF=9D?= =?UTF-8?q?=E9=95=BF=E4=BF=9D=E5=BA=8F=E5=89=8D=E6=8F=90=E4=B8=8B=E6=8C=89?= =?UTF-8?q?=E4=BD=8D=E7=AD=89=E4=BB=B7=E9=9B=B6=E6=88=90=E6=9C=AC=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- sanguo_factor/fast_ops.py | 38 ++++++++++++++++++++++++++++---------- 1 file changed, 28 insertions(+), 10 deletions(-) diff --git a/sanguo_factor/fast_ops.py b/sanguo_factor/fast_ops.py index b618049..7fc5ff9 100644 --- a/sanguo_factor/fast_ops.py +++ b/sanguo_factor/fast_ops.py @@ -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"),