Merge commit '15f327b8ae79d929638b4d15dafa45306833b7f9'

main/factor/concept_factor.py

@@ -0,0 +1,82 @@
+import pandas as pd
+import numpy as np
+from tqdm import tqdm
+
+def _prepare_concept_df(concept_dict: dict) -> pd.DataFrame:
+    """将 concept_dict 转换为长格式的 DataFrame。"""
+    records = []
+    for date_str, inner_dict in concept_dict.items():
+        trade_date = pd.to_datetime(date_str, format='%Y%m%d')
+        for concept_name, stock_list in inner_dict.items():
+            for ts_code in stock_list:
+                records.append((trade_date, concept_name, ts_code))
+    
+    if not records:
+        return pd.DataFrame(columns=['trade_date', 'concept_name', 'ts_code'])
+        
+    concept_df = pd.DataFrame(records, columns=['trade_date', 'concept_name', 'ts_code'])
+    concept_df = concept_df.drop_duplicates(subset=["trade_date", "ts_code"], keep="first")
+
+    return concept_df
+
+def generate_concept_factors(df: pd.DataFrame, concept_dict: dict) -> pd.DataFrame:
+    """
+    基于热门概念数据生成因子。
+
+    Args:
+        df (pd.DataFrame): 所有股票所有日期的数据，需包含 'ts_code', 'trade_date'
+                           以及用于聚合的列 (如 'pct_chg', 'turnover_rate')。
+        concept_dict (dict): 每日热门概念及其股票列表的字典。
+
+    Returns:
+        pd.DataFrame: 添加了概念相关因子的原始 DataFrame。
+    """
+    print("开始生成概念相关因子...")
+    
+    # 0. 准备工作，创建副本以避免修改原始df
+    df = df.copy()
+    df['trade_date'] = pd.to_datetime(df['trade_date'])
+
+    # 1. 将 concept_dict 转换为适合合并的 DataFrame
+    concept_df = _prepare_concept_df(concept_dict).sort_values(by=['trade_date'])
+    if concept_df.empty:
+        print("警告: concept_dict 为空或格式不正确，无法生成概念因子。")
+        return df
+
+    # 2. 将概念信息合并到主数据 df
+    df = pd.merge(df, concept_df, on=['trade_date', 'ts_code'], how='left')
+    # --- 因子计算 ---
+
+    # 因子 1: 是否属于当日热门概念
+    df['cat_hot_concept_stock'] = df['concept_name'].notna().astype(np.int8)
+
+    # 因子 2 & 3: 概念内的截面排序因子
+    # 创建一个掩码，只对热门概念股进行后续计算，以提高效率
+    hot_mask = df['concept_name'].notna()
+    
+    # 定义需要在概念内部进行截面排序的特征列表
+    # 确保这些列存在于你的 df 中
+    features_to_rank = ['pct_chg', 'turnover_rate', 'volume_ratio'] 
+    
+    # 筛选出 df 中实际存在的特征列
+    existing_features_to_rank = [f for f in features_to_rank if f in df.columns]
+    if not existing_features_to_rank:
+        print("警告: df 中缺少用于概念内排序的特征列，跳过相关因子计算。")
+    else:
+        print(f"开始计算概念内截面排序因子，基于: {existing_features_to_rank}")
+        
+        # 使用 groupby().rank() 高效计算截面排名
+        grouped = df[hot_mask].groupby(['trade_date', 'concept_name'])
+        
+        for feature in tqdm(existing_features_to_rank, desc="Ranking Features in Concepts"):
+            # 计算百分比排名 (0到1之间)，值越大表示排名越靠前
+            rank_col_name = f'concept_rank_{feature}'
+            df[rank_col_name] = grouped[feature].rank(pct=True)
+
+    # --- 清理 & 返回 ---
+    # `concept_name` 列包含了有用的信息，可以选择保留或删除
+    # 这里我们选择保留，以便后续分析。如果不需要，可以取消下面这行注释。
+    df.drop(columns=['concept_name'], inplace=True)
+    
+    print("概念相关因子生成完毕。")
+    return df

main/factor/factor.csv

@@ -1,63 +0,0 @@
-序号	因子名称 (Factor Name / Column Name)	因子类别 (Factor Category)	简要说明
-1	pe_ttm	价值类因子 (Value)	市盈率 TTM
-2	return_5, return_20	动量类因子 (Momentum)	过去5日/20日收益率
-3	act_factor1 to act_factor4	动量类 / 技术类因子 (Momentum / Technical)	基于不同周期EMA斜率计算的动量/趋势因子
-4	std_return_5, std_return_90, std_return_90_2	波动率类因子 (Volatility)	不同窗口期或延迟窗口期的滚动收益率标准差
-5	upside_vol, downside_vol	波动率类因子 (Volatility)	N日滚动上/下行波动率
-6	vol_ratio	波动率类因子 (Volatility)	上行波动率 / 下行波动率
-7	std_return_5 / std_return_90	波动率类因子 (Volatility)	短期波动率 / 长期波动率 比率
-8	std_return_90 - std_return_90_2	波动率类因子 (Volatility)	长期波动率与其10日前值的差值（波动变化）
-9	volatility (来自指数计算)	波动率类 / 市场因子 (Volatility / Market)	指数（或个股）的20日滚动收益率标准差
-10	log(circ_mv) (或 log_circ_mv)	市值类因子 (Size)	流通市值的对数值
-11	cs_rank_size	市值类因子 (Size)	对数流通市值的截面排序
-12	vol	流动性类因子 (Liquidity)	成交量 (通常需要与其他指标结合或处理)
-13	turnover_rate	流动性类因子 (Liquidity)	换手率
-14	volume_ratio	流动性类因子 (Liquidity)	量比
-15	turnover_deviation	流动性类因子 (Liquidity)	换手率与其3日滚动均值的标准差倍数偏离
-16	cat_turnover_spike	流动性类 / 分类因子 (Liquidity / Categorical)	换手率是否显著高于近期均值
-17	volume_change_rate	流动性类因子 (Liquidity)	短期滚动成交量均值 / 长期滚动成交量均值 - 1
-18	cat_volume_breakout	流动性类 / 分类因子 (Liquidity / Categorical)	当日成交量是否大于过去5日最大成交量
-19	avg_volume_ratio	流动性类因子 (Liquidity)	3日滚动量比均值
-20	cat_volume_ratio_breakout	流动性类 / 分类因子 (Liquidity / Categorical)	当日量比是否大于过去5日最大量比
-21	vol_spike (Rolling Mean Vol)	流动性类因子 (Liquidity)	20日滚动成交量均值
-22	vol_std_5	流动性类 / 波动率因子 (Liquidity / Volatility)	成交量日变化率的5日滚动标准差
-23	volume_growth	流动性类因子 (Liquidity)	20日成交量变化率
-24	turnover_std	流动性类 / 波动率因子 (Liquidity / Volatility)	换手率的20日滚动标准差
-25	flow_lg_elg_intensity	资金流 / 流动性类因子 (Money Flow / Liquidity)	（大单+超大单）净买入量 / 总成交量
-26	flow_divergence_diff, flow_divergence_ratio	资金流 / 情绪类因子 (Money Flow / Sentiment)	散户与主力资金流的差异或比率
-27	lg_elg_buy_prop	资金流 / 流动性类因子 (Money Flow / Liquidity)	（大单+超大单）买入量 / 总买入量
-28	flow_struct_buy_change	资金流 / 流动性类因子 (Money Flow / Liquidity)	主力买入占比的日变化
-29	flow_lg_elg_accel	资金流 / 动量类因子 (Money Flow / Momentum)	主力资金流加速度
-30	active_buy_volume_large/big/small	资金流 / 流动性类因子 (Money Flow / Liquidity)	不同规模主动买入量 / 净流入量
-31	buy_lg/elg_vol_minus_sell_lg/elg_vol	资金流 / 流动性类因子 (Money Flow / Liquidity)	不同规模净买入量 / 总净流入量
-32	cs_rank_net_lg_flow_val, cs_rank_elg_buy_ratio, cs_rank_lg_sm_flow_diverge, cs_rank_elg_buy_sell_sm_ratio	资金流 / 复合因子 (截面排序)	各种资金流指标的截面排序
-33	cs_rank_ind_adj_lg_flow	资金流 / 复合因子 (行业调整+截面排序)	行业调整后的大单净流入截面排序
-34	chip_concentration_range, chip_skewness, cost_support_15pct_change, weight_roc5, cost_stability, ctrl_strength, low_cost_dev, asymmetry, cost_conc_std_N, profit_pressure, underwater_resistance, cs_rank_rel_profit_margin, cs_rank_cost_breadth, cs_rank_dist_to_upper_cost	定位类因子 (Positioning) / 技术类	基于持仓成本分布 (cost_*, weight_avg) 计算的各种指标及其截面排序
-35	winner_rate, cs_rank_winner_rate	定位类因子 (Positioning) / 技术类	获利盘比例及其截面排序
-36	floating_chip_proxy, price_cost_divergence, high_cost_break_days, liquidity_risk, lock_factor, cost_atr_adj, smallcap_concentration, cat_golden_resonance	定位类因子 (Positioning) / 复合因子	结合持仓成本与其他信息（价格、成交、波动率、市值）的复合指标
-37	cat_winner_price_zone	定位类 / 分类因子 (Positioning / Categorical)	基于成本和获利盘划分的区域类别
-38	flow_chip_consistency, profit_taking_vs_absorb, vol_amp_loss, vol_drop_profit_cnt, cost_break_confirm_cnt, vol_wgt_hist_pos, cs_rank_vol_x_profit_margin, cs_rank_cost_dist_vol_ratio	定位类因子 (Positioning) / 复合因子	进一步结合定位、资金流、量价的复杂交互因子
-39	return_skew, return_kurtosis	技术类 / 统计特征 (Technical / Stats)	滚动收益率的偏度与峰度
-40	rsi_3	技术类 / 动量类因子 (Technical / Momentum)	3日相对强弱指数
-41	obv, maobv_6, obv-maobv_6	技术类 / 量价因子 (Technical / Volume)	能量潮及其均线、差离
-42	atr_14, atr_6	技术类 / 波动率类因子 (Technical / Volatility)	平均真实波幅
-43	log_close	技术类 / 量价因子 (Technical / Price)	收盘价对数
-44	up, down	技术类 / 量价因子 (Technical / Price Action)	标准化上影线、下影线长度
-45	alpha_22_improved, alpha_003, alpha_007, alpha_013	技术类 / Alpha因子 (Technical / Alpha)	WorldQuant Alpha 因子实现
-46	atr_norm_channel_pos	技术类 / 量价因子 (Technical / Price Action)	ATR 标准化的价格通道位置
-47	turnover_diff_skew	技术类 / 流动性类 (Technical / Liquidity)	换手率变化率的偏度
-48	pullback_strong_N_M	技术类 / 动量类因子 (Technical / Momentum)	近期强势股的回调幅度
-49	vol_adj_roc	技术类 / 复合因子 (动量+波动率)	波动率调整后的 N 日变化率
-50	ar, br, arbr	情绪类 / 技术类因子 (Sentiment / Technical)	ARBR 人气意愿指标
-51	up_ratio_20d (来自指数计算)	情绪类 / 市场因子 (Sentiment / Market)	指数（或个股）过去20天上涨天数比例
-52	cat_up_limit, cat_down_limit, up_limit_count_10d, down_limit_count_10d, consecutive_up_limit	事件驱动 / 市场状态因子 (Event / Market State)	涨跌停相关状态和计数
-53	momentum_factor, resonance_factor	复合因子 (量价) (Composite - P/V)	基于量、价、换手率等的简单复合
-54	cat_af2, cat_af3, cat_af4	复合因子 / 分类因子 (Composite / Cat.)	act_factor 之间的比较
-55	act_factor5, act_factor6	复合因子 (技术类) (Composite - Technical)	act_factor 1-4 的组合
-56	mv_volatility, mv_growth, mv_turnover_ratio, mv_adjusted_volume, mv_weighted_turnover, nonlinear_mv_volume, mv_volume_ratio, mv_momentum	复合因子 (市值+流动性/量价)	考虑了市值影响的量价、流动性或动量指标
-57	cap_neutral_cost_metric (占位符)	复合因子 / Alpha因子 (占位符)	市值行业中性化的成本指标（需实现）
-58	hurst_exponent_flow (占位符)	资金流 / 统计因子 (占位符)	资金流的 Hurst 指数（需实现）
-59	intraday_lg_flow_corr_N (占位符)	复合因子 (价格行为+资金流) (占位符)	日内趋势与大单流相关性（需实现）
-60	industry_* (来自 industry_df)	行业因子 (Industry)	对应行业的各种指标（如行业收益率、行业动量等）
-61	*_deviation (来自 create_deviation_within_dates)	复合因子 (相对行业)	个股因子相对于行业均值的偏离
-62	complex_factor_gplearn_1	复合因子 (GP生成)	DEAP/GP 找到的因子表达式 1

main/factor/money_factor.py

@@ -1,115 +1,112 @@
 import pandas as pd
 import numpy as np
 
-def holder_trade_factors(all_data_df: pd.DataFrame, stk_holdertrade_df: pd.DataFrame) -> pd.DataFrame:
+def holder_trade_factors(all_data_df: pd.DataFrame,
+                         stk_holdertrade_df: pd.DataFrame) -> pd.DataFrame:
     """
-    生成合并的股东增减持因子以及 change_ratio 相关因子（优化版）。
+    计算基于股东增减持数据的因子。
 
     Args:
-        stk_holdertrade_df (pd.DataFrame): 股东增减持数据，包含 'ts_code', 'ann_date', 'in_de', 'change_ratio'。
+        all_data_df (pd.DataFrame): 包含每日股票数据的 DataFrame，
-        all_data_df (pd.DataFrame): 所有日期所有股票数据，包含 'ts_code', 'trade_date'。
+                                    必须包含 'ts_code' 和 'trade_date' 列。
+        stk_holdertrade_df (pd.DataFrame): 包含股东增减持信息的 DataFrame。
+                                           必须包含 'ts_code', 'ann_date',
+                                           'in_de' (例如, '增持', '减持'),
+                                           和 'change_ratio'。
 
     Returns:
-        pd.DataFrame: 包含增减持因子的 all_data_df。
+        pd.DataFrame: 添加了新的股东增减持因子的 all_data_df DataFrame。
     """
-    print('正在计算股东增减持因子（优化版）...')
+    print("开始计算股东增减持因子...")
 
-    # 1. 确保日期列为 datetime 类型
+    # --- 1. 预处理 stk_holdertrade_df ---
-    stk_holdertrade_df['ann_date'] = pd.to_datetime(stk_holdertrade_df['ann_date'])
+    # 创建副本以避免修改原始传入的DataFrame
-    all_data_df['trade_date'] = pd.to_datetime(all_data_df['trade_date'])
+    stk_trade_processed_df = stk_holdertrade_df.copy()
 
-    # 2. 对增减持数据进行预处理和排序（排序在此阶段可能不是严格必需的，但保持良好习惯）
+    # 确保 'ann_date' 是 datetime 类型
-    holder_data_processed = stk_holdertrade_df.copy()
+    stk_trade_processed_df['ann_date'] = pd.to_datetime(stk_trade_processed_df['ann_date'])
-    holder_data_processed['change_ratio_in_agg'] = holder_data_processed['change_ratio'].where(holder_data_processed['in_de'] == 'IN', 0)
-    holder_data_processed['change_ratio_de_agg'] = holder_data_processed['change_ratio'].where(holder_data_processed['in_de'] == 'DE', 0)
-    holder_data_processed['change_ratio_total_agg'] = holder_data_processed['change_ratio']
-    holder_data_processed['in_de_numeric'] = holder_data_processed['in_de'].map({'IN': 1, 'DE': -1}).fillna(0) # 用于判断类型
 
-    # 提前获取所有唯一的交易日期集合，以提高查找效率
+    # 将 'in_de' 映射为数值: 1 代表 '增持', -1 代表 '减持'
-    all_trade_dates_set = set(all_data_df['trade_date'].unique())
+    # 请根据你数据中实际的 'in_de' 字符串调整
+    in_de_map = {'增持': 1, '减持': -1} # 假设是这两个值
+    # 如果你的值是 '1' 和 '2' (1代表增持, 2代表减持)，则映射应相应调整
+    # 或者如果 'in_de' 已经是 1 和 -1 (或类似数值)，则可以跳过映射，但要确保类型正确
+    stk_trade_processed_df['_direction'] = stk_trade_processed_df['in_de'].map(in_de_map)
+    # 如果 _direction 列在映射后可能产生NaN (因为in_de中有未覆盖的值)，需要处理
+    if stk_trade_processed_df['_direction'].isnull().any():
+        print("警告: 'in_de' 列中存在未映射的值，可能导致 _direction 列出现NaN。")
+        # 可以选择填充NaN，例如用0填充，或者移除这些行
+        # stk_trade_processed_df['_direction'].fillna(0, inplace=True)
 
-    # 3. 构建一个辅助DataFrame，记录每个公告在未来10个日历日（且是交易日）的影响
+    # 计算有效变动比例 (方向 * 变动比例)
-    expanded_holder_events = []
+    # 确保 change_ratio 是数值类型。假设 change_ratio 是一个正确的比例值 (例如 0.005 代表 0.5%)。
-    for _, row in holder_data_processed.iterrows():
+    # 如果 change_ratio 是百分点 (例如 0.5 代表 0.5%)，你可能需要除以 100。
-        ts_code = row['ts_code']
+    stk_trade_processed_df['change_ratio'] = pd.to_numeric(stk_trade_processed_df['change_ratio'], errors='coerce')
-        ann_date = row['ann_date']
+    stk_trade_processed_df['_effective_change'] = stk_trade_processed_df['_direction'] * stk_trade_processed_df['change_ratio']
-        
-        # 生成从公告日期开始的未来10个日历日的日期范围（包括公告日本身）
-        # pd.Timedelta(days=10) 表示从公告日+10天
-        # pd.date_range(start=ann_date, end=ann_date + pd.Timedelta(days=10), freq='D')
-        # 更精确地生成11个日期，涵盖公告日及其后的10个日历日
-        future_dates = pd.date_range(start=ann_date, periods=11, freq='D') 
 
-        for date_in_window in future_dates:
+    # 按股票代码和公告日期聚合当日的多次增减持操作
-            # 只有当日期是实际交易日时才添加
+    daily_trade_agg = stk_trade_processed_df.groupby(['ts_code', 'ann_date']).agg(
-            if date_in_window in all_trade_dates_set:
+        net_change_ratio_daily=('_effective_change', 'sum'),
-                expanded_holder_events.append({
+        any_increase_daily=('_direction', lambda x: (x == 1).any().astype(int)),
-                    'ts_code': ts_code,
+        any_decrease_daily=('_direction', lambda x: (x == -1).any().astype(int))
-                    'trade_date': date_in_window,
-                    'in_de_numeric': row['in_de_numeric'],
-                    'change_ratio_total_agg': row['change_ratio_total_agg'],
-                    'change_ratio_in_agg': row['change_ratio_in_agg'],
-                    'change_ratio_de_agg': row['change_ratio_de_agg']
-                })
-    
-    if not expanded_holder_events: # 如果没有事件，直接返回原始 df
-        # 确保返回的DataFrame与原始df具有相同的列和顺序
-        # 并填充为默认值
-        default_factors = pd.DataFrame({
-            'holder_trade_type_10d': None,
-            'holder_change_ratio_sum_10d': 0.0,
-            'holder_in_change_ratio_sum_10d': 0.0,
-            'holder_de_change_ratio_sum_10d': 0.0
-        }, index=all_data_df.index)
-        return pd.concat([all_data_df, default_factors], axis=1)
-
-
-    expanded_holder_events_df = pd.DataFrame(expanded_holder_events)
-
-    # 4. 聚合每个 (ts_code, trade_date) 对上的事件
-    # 可能会有重复的 (ts_code, trade_date) 对，因为一个交易日可能受多个公告影响
-    daily_aggregated_factors = expanded_holder_events_df.groupby(['ts_code', 'trade_date']).agg(
-        holder_change_ratio_sum_10d=('change_ratio_total_agg', 'sum'),
-        holder_in_change_ratio_sum_10d=('change_ratio_in_agg', 'sum'),
-        holder_de_change_ratio_sum_10d=('change_ratio_de_agg', 'sum'),
-        # 对于 holder_trade_type_10d，聚合 in_de_numeric 的唯一值集合
-        _in_de_types_unique=('in_de_numeric', lambda x: set(x)) # 获取该日期窗口内所有独特的增减持类型
     ).reset_index()
 
-    # 根据 _in_de_types_unique 确定 holder_trade_type_10d
+    # 将 'ann_date' 重命名为 'trade_date' 以便与 all_data_df 合并
-    def get_trade_type(unique_types_set):
+    daily_trade_agg = daily_trade_agg.rename(columns={'ann_date': 'trade_date'})
-        if 1 in unique_types_set and -1 in unique_types_set:
-            return 'BOTH'
-        elif 1 in unique_types_set:
-            return 'IN'
-        elif -1 in unique_types_set:
-            return 'DE'
-        else:
-            return None # 理论上不应该发生，除非 unique_types_set 为空或只包含0
-            
-    daily_aggregated_factors['holder_trade_type_10d'] = daily_aggregated_factors['_in_de_types_unique'].apply(get_trade_type)
 
-    # 移除辅助列
+    # --- 2. 与 all_data_df 合并 ---
-    daily_aggregated_factors.drop(columns=['_in_de_types_unique'], inplace=True)
+    # 创建 all_data_df 的副本进行操作
+    df_merged = all_data_df.copy()
+    df_merged['trade_date'] = pd.to_datetime(df_merged['trade_date']) # 确保日期类型一致
 
-    # 5. 将计算得到的因子合并回 all_data_df
+    # 使用左合并保留 all_data_df 的所有行，并在有增减持公告的日期添加信息
-    # 确保 all_data_df 也按 ts_code, trade_date 排序，以便 merge 高效
+    df_merged = pd.merge(df_merged, daily_trade_agg, on=['ts_code', 'trade_date'], how='left')
-    all_data_df_sorted = all_data_df.sort_values(['ts_code', 'trade_date']).reset_index(drop=True)
-    
-    final_df = pd.merge(
-        all_data_df_sorted,
-        daily_aggregated_factors,
-        on=['ts_code', 'trade_date'],
-        how='left'
-    )
 
-    # 6. 对于没有增减持记录的日期，因子值为 None 或 0
+    # 对于没有增减持公告的日期，填充NaN值为0（表示当天无变动或无公告）
-    # 在 merge_asof 中确实需要排序，但在这种事件展开的方法中，merge 是普通的 left merge，不需要预排序。
+    df_merged['net_change_ratio_daily'] = df_merged['net_change_ratio_daily'].fillna(0)
-    # 考虑到最终的 merge，最好还是保持 `all_data_df` 和 `daily_aggregated_factors` 的键排序。
+    df_merged['any_increase_daily'] = df_merged['any_increase_daily'].fillna(0)
-    # 所以在 `merge` 前对 `all_data_df` 进行一次排序是好的实践。
+    df_merged['any_decrease_daily'] = df_merged['any_decrease_daily'].fillna(0)
-    
+
-    final_df['holder_trade_type_10d'] = final_df['holder_trade_type_10d'].fillna(None)
+    # --- 3. 计算滚动因子 ---
-    fillna_ratio_cols = ['holder_change_ratio_sum_10d', 'holder_in_change_ratio_sum_10d', 'holder_de_change_ratio_sum_10d']
+    # !!! 关键步骤：确保在 groupby().rolling() 之前按分组键和时间键排序 !!!
-    final_df[fillna_ratio_cols] = final_df[fillna_ratio_cols].fillna(0.0)
+    # 这一步至关重要，以保证滚动计算后的 Series 在 reset_index 后能正确对齐
+    df_merged = df_merged.sort_values(['ts_code', 'trade_date']).reset_index(drop=True)
+
+    grouped = df_merged.groupby('ts_code', group_keys=False) # group_keys=False 避免在结果中保留分组键作为索引层级
+
+    # 因子: 过去N日净变动比例之和
+    for N in [10]:
+        rolling_series = grouped['net_change_ratio_daily'].rolling(window=N, min_periods=1).sum()
+        df_merged[f'holder_net_change_sum_{N}d'] = rolling_series.reset_index(level=0, drop=True)
+
+    # 因子: 过去N日发生增持的天数
+    for N in [10]:
+        rolling_series = grouped['any_increase_daily'].rolling(window=N, min_periods=1).sum()
+        df_merged[f'holder_increase_days_{N}d'] = rolling_series.reset_index(level=0, drop=True)
+
+    # 因子: 过去N日发生减持的天数
+    for N in [10]:
+        rolling_series = grouped['any_decrease_daily'].rolling(window=N, min_periods=1).sum()
+        df_merged[f'holder_decrease_days_{N}d'] = rolling_series.reset_index(level=0, drop=True)
+
+    # 因子: 过去N日是否发生过增持 (布尔标志)
+    for N in [10]:
+        rolling_series_sum = grouped['any_increase_daily'].rolling(window=N, min_periods=1).sum()
+        df_merged[f'holder_any_increase_flag_{N}d'] = (rolling_series_sum > 0).astype(int).reset_index(level=0, drop=True)
+
+    # 因子: 过去N日是否发生过减持 (布尔标志)
+    for N in [10]:
+        rolling_series_sum = grouped['any_decrease_daily'].rolling(window=N, min_periods=1).sum()
+        df_merged[f'holder_any_decrease_flag_{N}d'] = (rolling_series_sum > 0).astype(int).reset_index(level=0, drop=True)
+
+    # 因子: 过去N日净“活动”得分 (增持天数 - 减持天数)
+    for N in [10]:
+        # 直接使用已经计算好且索引对齐的列
+        df_merged[f'holder_direction_score_{N}d'] = df_merged[f'holder_increase_days_{N}d'] - df_merged[f'holder_decrease_days_{N}d']
+
+    # 清理在合并过程中产生的每日临时列（如果不再需要它们）
+    df_merged.drop(columns=['net_change_ratio_daily', 'any_increase_daily', 'any_decrease_daily'], inplace=True, errors='ignore')
+
+    print("股东增减持因子计算完成。")
+    return df_merged
 
-    return final_df

main/utils/data_process.py

@@ -0,0 +1,268 @@
+import pandas as pd
+import numpy as np
+import statsmodels.api as sm # 用于中性化回归
+from tqdm import tqdm
+
+epsilon = 1e-10 # 防止除零
+
+def zscore_standardize(train_df: pd.DataFrame, test_df: pd.DataFrame, features: list, epsilon: float = 1e-10):
+    """
+    对指定特征列进行截面 Z-Score 标准化 (原地修改)。
+    方法: Z = (value - cross_sectional_mean) / (cross_sectional_std + epsilon)
+
+    Args:
+        df (pd.DataFrame): 输入 DataFrame，需包含 'trade_date' 和 features 列。
+        features (list): 需要处理的特征列名列表。
+        epsilon (float): 防止除以零的小常数。
+
+    WARNING: 此函数会原地修改输入的 DataFrame 'df'。
+    """
+    print("开始截面 Z-Score 标准化...")
+    if not all(col in train_df.columns for col in features):
+        missing = [col for col in features if col not in train_df.columns]
+        print(f"错误: DataFrame 中缺少以下特征列: {missing}。跳过标准化处理。")
+        return
+
+
+    for col in tqdm(features, desc="Standardizing"):
+        try:
+            # 使用 transform 计算截面均值和标准差
+            mean = train_df[col].transform('mean')
+            std = train_df[col].transform('std')
+
+            # 计算 Z-Score 并原地赋值
+            train_df[col] = (train_df[col] - mean) / (std + epsilon)
+            test_df[col] = (test_df[col] - mean) / (std + epsilon)
+
+        except KeyError:
+            print(f"警告: 列 '{col}' 可能不存在或在分组中出错，跳过此列的标准化处理。")
+        except Exception as e:
+            print(f"警告: 处理列 '{col}' 时发生错误: {e}，跳过此列的标准化处理。")
+
+    print("截面 Z-Score 标准化完成。")
+
+
+# --- 1. 中位数去极值 (MAD) ---
+
+def cs_mad_filter(df: pd.DataFrame,
+                  features: list,
+                  k: float = 3.0,
+                  scale_factor: float = 1.4826):
+    """
+    对指定特征列进行截面 MAD 去极值处理 (原地修改)。
+
+    方法: 对每日截面数据，计算 median 和 MAD，
+          将超出 [median - k * scale * MAD, median + k * scale * MAD] 范围的值
+          替换为边界值 (Winsorization)。
+          scale_factor=1.4826 使得 MAD 约等于正态分布的标准差。
+
+    Args:
+        df (pd.DataFrame): 输入 DataFrame，需包含 'trade_date' 和 features 列。
+        features (list): 需要处理的特征列名列表。
+        k (float): MAD 的倍数，用于确定边界。默认为 3.0。
+        scale_factor (float): MAD 的缩放因子。默认为 1.4826。
+
+    WARNING: 此函数会原地修改输入的 DataFrame 'df'。
+    """
+    print(f"开始截面 MAD 去极值处理 (k={k})...")
+    if not all(col in df.columns for col in features):
+        missing = [col for col in features if col not in df.columns]
+        print(f"错误: DataFrame 中缺少以下特征列: {missing}。跳过去极值处理。")
+        return
+
+    grouped = df.groupby('trade_date')
+
+    for col in tqdm(features, desc="MAD Filtering"):
+        try:
+            # 计算截面中位数
+            median = grouped[col].transform('median')
+            # 计算截面 MAD (Median Absolute Deviation from Median)
+            mad = (df[col] - median).abs().groupby(df['trade_date']).transform('median')
+
+            # 计算上下边界
+            lower_bound = median - k * scale_factor * mad
+            upper_bound = median + k * scale_factor * mad
+
+            # 原地应用 clip
+            df[col] = np.clip(df[col], lower_bound, upper_bound)
+
+        except KeyError:
+            print(f"警告: 列 '{col}' 可能不存在或在分组中出错，跳过此列的 MAD 处理。")
+        except Exception as e:
+            print(f"警告: 处理列 '{col}' 时发生错误: {e}，跳过此列的 MAD 处理。")
+
+    print("截面 MAD 去极值处理完成。")
+
+
+# --- 2. 行业市值中性化 ---
+
+def cs_neutralize_industry_cap(df: pd.DataFrame,
+                               features: list,
+                               industry_col: str = 'cat_l2_code',
+                               market_cap_col: str = 'circ_mv'):
+    """
+    对指定特征列进行截面行业和对数市值中性化 (原地修改)。
+    使用 OLS 回归: feature ~ 1 + log(market_cap) + C(industry)
+    将回归残差写回原特征列。
+
+    Args:
+        df (pd.DataFrame): 输入 DataFrame，需包含 'trade_date', features 列,
+                           industry_col, market_cap_col。
+        features (list): 需要处理的特征列名列表。
+        industry_col (str): 行业分类列名。
+        market_cap_col (str): 流通市值列名。
+
+    WARNING: 此函数会原地修改输入的 DataFrame 'df' 的 features 列。
+             计算量较大，可能耗时较长。
+             需要安装 statsmodels 库 (pip install statsmodels)。
+    """
+    print("开始截面行业市值中性化...")
+    required_cols = features + ['trade_date', industry_col, market_cap_col]
+    if not all(col in df.columns for col in required_cols):
+        missing = [col for col in required_cols if col not in df.columns]
+        print(f"错误: DataFrame 中缺少必需列: {missing}。无法进行中性化。")
+        return
+
+    # 预处理：计算 log 市值，处理 industry code 可能的 NaN
+    log_cap_col = '_log_market_cap'
+    df[log_cap_col] = np.log1p(df[market_cap_col]) # log1p 处理 0 值
+    # df[industry_col] = df[industry_col].cat.add_categories('UnknownIndustry')
+    # df[industry_col] = df[industry_col].fillna('UnknownIndustry') # 填充行业 NaN
+    # df[industry_col] = df[industry_col].astype('category') # 转为类别，ols 会自动处理
+
+    dates = df['trade_date'].unique()
+    all_residuals = [] # 用于收集所有日期的残差
+
+    for date in tqdm(dates, desc="Neutralizing"):
+        daily_data = df.loc[df['trade_date'] == date, features + [log_cap_col, industry_col]].copy() # 使用 .loc 获取副本
+
+        # 准备自变量 X (常数项 + log市值 + 行业哑变量)
+        X = daily_data[[log_cap_col]]
+        X = sm.add_constant(X, prepend=True) # 添加常数项
+        # 创建行业哑变量 (drop_first=True 避免共线性)
+        industry_dummies = pd.get_dummies(daily_data[industry_col], prefix=industry_col, drop_first=True)
+        industry_dummies = industry_dummies.astype(int)
+        X = pd.concat([X, industry_dummies], axis=1)
+
+        daily_residuals = daily_data[[col for col in features]].copy() # 创建用于存储残差的df
+
+        for col in features:
+            Y = daily_data[col]
+
+            # 处理 NaN 值，确保 X 和 Y 在相同位置有有效值
+            valid_mask = Y.notna() & X.notna().all(axis=1)
+            if valid_mask.sum() < (X.shape[1] + 1): # 数据点不足以估计模型
+                print(f"警告: 日期 {date}, 特征 {col} 有效数据不足 ({valid_mask.sum()}个)，无法中性化，填充 NaN。")
+                daily_residuals[col] = np.nan
+                continue
+
+            Y_valid = Y[valid_mask]
+            X_valid = X[valid_mask]
+
+            # 执行 OLS 回归
+            try:
+                model = sm.OLS(Y_valid.to_numpy(), X_valid.to_numpy())
+                results = model.fit()
+                # 将残差填回对应位置
+                daily_residuals.loc[valid_mask, col] = results.resid
+                daily_residuals.loc[~valid_mask, col] = np.nan # 原本无效的位置填充 NaN
+            except Exception as e:
+                print(f"警告: 日期 {date}, 特征 {col} 回归失败: {e}，填充 NaN。")
+                daily_residuals[col] = np.nan
+                break
+
+        all_residuals.append(daily_residuals)
+
+    # 合并所有日期的残差结果
+    if all_residuals:
+        residuals_df = pd.concat(all_residuals)
+        # 将残差结果更新回原始 df (原地修改)
+        # 使用 update 比 merge 更适合基于索引的原地更新
+        # 确保 residuals_df 的索引与 df 中对应部分一致
+        df.update(residuals_df)
+    else:
+        print("没有有效的残差结果可以合并。")
+
+
+    # 清理临时列
+    df.drop(columns=[log_cap_col], inplace=True)
+    print("截面行业市值中性化完成。")
+
+
+# --- 3. Z-Score 标准化 ---
+
+def cs_zscore_standardize(df: pd.DataFrame, features: list, epsilon: float = 1e-10):
+    """
+    对指定特征列进行截面 Z-Score 标准化 (原地修改)。
+    方法: Z = (value - cross_sectional_mean) / (cross_sectional_std + epsilon)
+
+    Args:
+        df (pd.DataFrame): 输入 DataFrame，需包含 'trade_date' 和 features 列。
+        features (list): 需要处理的特征列名列表。
+        epsilon (float): 防止除以零的小常数。
+
+    WARNING: 此函数会原地修改输入的 DataFrame 'df'。
+    """
+    print("开始截面 Z-Score 标准化...")
+    if not all(col in df.columns for col in features):
+        missing = [col for col in features if col not in df.columns]
+        print(f"错误: DataFrame 中缺少以下特征列: {missing}。跳过标准化处理。")
+        return
+
+    grouped = df.groupby('trade_date')
+
+    for col in tqdm(features, desc="Standardizing"):
+        try:
+            # 使用 transform 计算截面均值和标准差
+            mean = grouped[col].transform('mean')
+            std = grouped[col].transform('std')
+
+            # 计算 Z-Score 并原地赋值
+            df[col] = (df[col] - mean) / (std + epsilon)
+
+        except KeyError:
+            print(f"警告: 列 '{col}' 可能不存在或在分组中出错，跳过此列的标准化处理。")
+        except Exception as e:
+            print(f"警告: 处理列 '{col}' 时发生错误: {e}，跳过此列的标准化处理。")
+
+    print("截面 Z-Score 标准化完成。")
+
+def fill_nan_with_daily_median(df: pd.DataFrame, feature_columns: list[str]) -> pd.DataFrame:
+    """
+    对指定特征列进行每日截面中位数填充缺失值 (NaN)。
+
+    参数:
+    df (pd.DataFrame): 包含多日数据的DataFrame，需要包含 'trade_date' 和 feature_columns 中的列。
+    feature_columns (list[str]): 需要进行缺失值填充的特征列名称列表。
+
+    返回:
+    pd.DataFrame: 包含缺失值填充后特征列的DataFrame。在输入DataFrame的副本上操作。
+    """
+    processed_df = df.copy() # 在副本上操作，保留原始数据
+
+    # 确保 trade_date 是 datetime 类型以便正确分组
+    processed_df['trade_date'] = pd.to_datetime(processed_df['trade_date'])
+
+    def _fill_daily_nan(group):
+        # group 是某一个交易日的 DataFrame
+
+        # 遍历指定的特征列
+        for feature_col in feature_columns:
+            # 检查列是否存在于当前分组中
+            if feature_col in group.columns:
+                # 计算当日该特征的中位数
+                median_val = group[feature_col].median()
+
+                # 使用当日中位数填充该特征列的 NaN 值
+                # inplace=True 会直接修改 group DataFrame
+                group[feature_col].fillna(median_val, inplace=True)
+            # else:
+            #     print(f"Warning: Feature column '{feature_col}' not found in daily group for {group['trade_date'].iloc[0]}. Skipping.")
+
+        return group
+
+    # 按交易日期分组，并应用每日填充函数
+    # group_keys=False 避免将分组键添加到结果索引中
+    filled_df = processed_df.groupby('trade_date', group_keys=False).apply(_fill_daily_nan)
+
+    return filled_df

main/utils/indicators.py

@@ -0,0 +1,130 @@
+import pandas as pd
+import numpy as np
+
+def calculate_sharpe_sortino_labels_efficient(
+    df: pd.DataFrame,
+    N_days: int = 5,  # 未来考察的天数，用于计算收益率序列
+    risk_free_rate_annual: float = 0.02, # 年化无风险利率 (例如 0.02 表示 2%)
+    annualization_factor: float = np.sqrt(252), # 年化因子，日数据通常用sqrt(252)
+    min_periods_for_std: int = 2 # 计算标准差所需的最小有效数据点数
+) -> pd.DataFrame:
+    """
+    高效计算每只股票在每个交易日未来 N 日的年化夏普比率和索提诺比率，作为训练的Label。
+    函数会获取未来 N 天的每日收益率序列，并基于此序列计算夏普/索提诺。
+
+    Args:
+        df (pd.DataFrame): 输入 DataFrame，需包含 'ts_code', 'trade_date', 'close' 列。
+                           'close' 列必须是股票收盘价。
+        N_days (int): 未来考察的天数。例如，N_days=5 表示考察未来5个交易日。
+        risk_free_rate_annual (float): 年化无风险利率。
+        annualization_factor (float): 用于将日收益率转化为年化收益率的因子 (例如 np.sqrt(252))。
+        min_periods_for_std (int): 计算标准差所需的最小有效数据点数。
+                                   如果未来N天有效数据少于此值，则 Label 为 NaN。
+
+    Returns:
+        pd.DataFrame: 原始DataFrame，新增 'sharpe_ratio_label' 和 'sortino_ratio_label' 列。
+                      注意：数据末尾 N_days 行的 Label 将为 NaN。
+    """
+    df_copy = df
+    
+    # 确保数据已按股票和日期排序，这对于高效计算至关重要
+    df_copy = df_copy.sort_values(by=['ts_code', 'trade_date'])
+
+    # # 计算每日收益率
+    # df_copy['pct_chg'] = df_copy.groupby('ts_code')['close'].pct_change()
+
+    # 将年化无风险利率转换为每日无风险利率
+    daily_risk_free_rate = risk_free_rate_annual / 252 # 假设每年252个交易日
+
+    def _calculate_metrics_for_window(returns_series):
+        """
+        辅助函数：计算单个滚动窗口内的夏普和索提诺比率。
+        这个函数将被 apply 调用于每个 (N_days) 收益率序列。
+        """
+        if len(returns_series.dropna()) < min_periods_for_std:
+            return np.nan, np.nan # 数据不足，返回NaN
+
+        excess_returns = returns_series - daily_risk_free_rate
+
+        # --- 夏普比率 ---
+        mean_excess_return = excess_returns.mean()
+        std_dev_returns = returns_series.std() # 夏普比率使用总收益率的标准差
+
+        sharpe = np.nan
+        if std_dev_returns != 0:
+            sharpe = (mean_excess_return / std_dev_returns) * annualization_factor
+
+        # --- 索提诺比率 ---
+        downside_returns = returns_series[returns_series < daily_risk_free_rate]
+        sortino = np.nan
+        if not downside_returns.empty:
+            downside_deviation = np.sqrt(np.mean((downside_returns - daily_risk_free_rate)**2))
+            if downside_deviation != 0:
+                sortino = (mean_excess_return / downside_deviation) * annualization_factor
+
+        return sharpe, sortino
+
+    # 使用 groupby().rolling() 结合 apply 来高效计算
+    # 注意: 这里使用 shift(-N_days) 来获取未来 N_days 的窗口
+    # 同时 rolling 窗口需要是 N_days 大小，并对 pct_chg 序列进行操作
+    # 结果会被放置在窗口的末尾，但我们实际上需要它对应到窗口的起始位置
+    # 因此，我们先计算，然后将其向上 shift N_days + 1 （或者根据 rolling 的行为调整）
+    
+    # 获取未来 N_days 的每日收益率序列
+    # 为了避免在每一行中循环，我们构造一个辅助的DataFrame，其中包含未来N天的收益率
+    # 这里使用一个更高效的思路：对每只股票，生成一个N天的收益率列表
+    
+    # 使用 groupby().apply() 和 list comprehension 来构建未来N天的收益率列表
+    # 为了获得未来N天的收益率，需要对pct_chg进行shift(-N_days)
+    # 然后再对每个N_days的窗口进行处理
+    
+    # 方法一：先获取未来N天的收益率列，然后进行滑动窗口计算 （更推荐，效率高）
+    # 创建一个辅助列，表示未来 N 天的每日收益率列表
+    # 注意：这里需要 N_days 长度的窗口，且数据点至少为 min_periods_for_std
+    
+    # 对每只股票进行分组，然后计算未来 N 天的夏普/索提诺
+    # 这仍然涉及到对每个时间点获取未来N天的收益序列，Pandas 的 rolling 默认是向后看的
+    # 为了实现向后看 N_days (即从当前日期 t 预测 t+1 到 t+N_days)，我们需要一些技巧
+    
+    # 最直接且通常高效的方法是：
+    # 1. 计算日收益率。
+    # 2. 对每个股票，对日收益率进行反向滚动，然后计算指标。
+    #    或者更简单地，使用 shift(-N_days) 创建“未来的”日收益率列，然后进行滚动。
+    
+    # 为了简化且保持效率，我们可以这样做：
+    # 1. 计算每个股票的每日收益率。
+    # 2. 对于每个 (ts_code, trade_date) 行，我们将查看其未来的 N_days 每日收益率。
+    #    这在 Pandas 中不直接通过 .rolling() 实现，因为 rolling 是“当前及之前”的窗口。
+    #    我们可以用 groupby().apply() 配合自定义函数来模拟这个“未来窗口”的行为。
+
+    def _apply_metrics_per_stock(stock_df_group):
+        """
+        对单只股票的数据帧进行处理，计算其每个日期的未来 N 日夏普/索提诺比率。
+        """
+        sharpe_labels = [np.nan] * len(stock_df_group)
+        sortino_labels = [np.nan] * len(stock_df_group)
+
+        # 遍历当前股票的每个日期，获取未来的 N_days 收益率序列
+        for i in range(len(stock_df_group) - N_days):
+            # 获取未来 N_days 的每日收益率序列
+            # 从当前日期 t 的下一天 (t+1) 开始，到 t+N_days
+            future_pct_chgs = stock_df_group['pct_chg'].iloc[i+1 : i+1+N_days].dropna()
+            
+            # 计算夏普和索提诺
+            sharpe, sortino = _calculate_metrics_for_window(future_pct_chgs)
+            
+            # 赋值到对应日期
+            sharpe_labels[i] = sharpe
+            sortino_labels[i] = sortino
+        
+        stock_df_group['sharpe_ratio_label'] = sharpe_labels
+        stock_df_group['sortino_ratio_label'] = sortino_labels
+        return stock_df_group
+
+    # 对每个股票分组应用这个函数
+    df_result = df_copy.groupby('ts_code', group_keys=False).apply(_apply_metrics_per_stock)
+
+    # # 清理不再需要的中间列
+    # df_result = df_result.drop(columns=['pct_chg'])
+
+    return df_result