双策略A股选股器

-- coding: utf-8 --

""" 双策略选股器 - 通用版 策略1: 仙人指路B + 持股20天 策略2: 老鸭头B + 持股5天

使用方法: python dual_strategy_selector.py --data your_data.csv python dual_strategy_selector.py --data your_data.csv --date 2026-04-30 python dual_strategy_selector.py --data your_data.csv --watch

数据接口要求: CSV文件必须包含以下列:

symbol: 股票代码 name: 股票名称 date: 日期 (YYYY-MM-DD) open: 开盘价 high: 最高价 low: 最低价 close: 收盘价 volume: 成交量 turnover: 换手率 (%) ma5: 5日均线 ma10: 10日均线 ma20: 20日均线 ma60: 60日均线 volume_ratio: 量比 pct_change: 涨跌幅 (%) is_up_limit: 是否涨停 (0/1) """

导入 pandas as pd 导入 numpy as np from datetime 导入 datetime 导入 arg解析 导入 os

============================================================ 策略配置 ============================================================

STRATEGY_CONFIG = { '仙人指路B': { 'name': '仙人指路B', 'version': '1.0', 'hold_days': 20, 'expected_win_rate': 54.2, 'expected_avg_profit': 3.53, 'params': { 'upper_shadow_pct': 2.0, # 上影线>2% 'body_pct_max': 2.0, # 实体<2% 'turnover_min': 3.0, # 换手>3% 'turnover_max': 7.0, # 换手<7% 'volume_ratio_min': 1.0, # 量比>1 } }, '老鸭头B': { 'name': '老鸭头B', 'version': '1.0', 'hold_days': 5, 'expected_win_rate': 72.0, 'expected_avg_profit': 0.77, 'params': { 'near_pct': 0.02, # 回调至20日线±2% 'require_10x60': True, # 10日上穿60日 'require_ma10_up': True, # MA10向上 'require_price_above_20': False, } } }

============================================================ 数据加载函数 ============================================================

def load_data(data_path, tar获取_date=None): """ 加载用户数据

参数: data_path: CSV文件路径 tar获取_date: 目标日期 (可选)

返回: DataFrame """ print(f"[OK] 加载数据: {data_path}")

if not os.path.exists(data_path): print(f"[ERROR] 文件不存在: {data_path}") return None

df = pd.read_csv(data_path)

# 检查必要列 required_cols = ['symbol', 'name', 'date', 'open', 'high', 'low', 'close', 'volume', 'turnover', 'ma5', 'ma10', 'ma20', 'ma60', 'volume_ratio', 'pct_change', 'is_up_limit']

missing_cols = [col for col in required_cols if col not in df.columns] if missing_cols: print(f"[ERROR] 缺少必要列: {missing_cols}") print("[信息] 请确保CSV包含以下列:") for col in required_cols: print(f" - {col}") return None

# 转换日期 df['date'] = pd.to_datetime(df['date'])

print(f"[OK] 数据: {len(df)}行, {df['symbol'].nunique()}只股票") print(f"[OK] 日期范围: {df['date'].min()} ~ {df['date'].max()}")

return df

============================================================ 信号检测函数 ============================================================

def 检测_签名als(df): """检测双策略信号""" df = df.排序_values(['symbol', 'date'])

# 预计算shift df['ma5_prev'] = df.groupby('symbol')['ma5'].shift(1) df['ma10_prev'] = df.groupby('symbol')['ma10'].shift(1) df['ma20_prev'] = df.groupby('symbol')['ma20'].shift(1) df['ma60_prev'] = df.groupby('symbol')['ma60'].shift(1) df['close_prev'] = df.groupby('symbol')['close'].shift(1)

# === 仙人指路B === df['body'] = abs(df['close'] - df['open']) df['body_pct'] = df['body'] / df['close_prev'] 100 df['upper_shadow'] = df['high'] - df[['open', 'close']].max(axis=1) df['upper_shadow_pct'] = df['upper_shadow'] / df['close_prev'] 100

params_xr = STRATEGY_CONFIG['仙人指路B']['params'] df['签名al_xrB'] = ( (df['upper_shadow_pct'] > params_xr['upper_shadow_pct']) & (df['body_pct'] < params_xr['body_pct_max']) & (df['turnover'] >= params_xr['turnover_min']) & (df['turnover'] <= params_xr['turnover_max']) & (df['volume_ratio'] > params_xr['volume_ratio_min']) ).astype(int)

# === 老鸭头B === df['s_5x10'] = ((df['ma5_prev'] < df['ma10_prev']) & (df['ma5'] >= df['ma10'])).astype(int) df['ma20_slope'] = df.groupby('symbol')['ma20'].转换(lambda x: x.diff().rolling(5).mean()) df['s_20_up'] = (df['ma20_slope'] > 0).astype(int) df['price_near_20'] = (abs(df['close'] - df['ma20']) / df['ma20'] < STRATEGY_CONFIG['老鸭头B']['params']['near_pct']).astype(int) df['签名al_lyB_base'] = (df['s_5x10'] & df['s_20_up'] & df['price_near_20']).astype(int)

# 加成条件 df['s_10x60'] = ((df['ma10_prev'] < df['ma60_prev']) & (df['ma10'] >= df['ma60'])).astype(int) df['s_ma10_up'] = (df['ma10'] > df['ma10_prev']).astype(int) df['s_price_above_20'] = (df['close'] > df['ma20']).astype(int)

# 老鸭头B完整信号 ly_params = STRATEGY_CONFIG['老鸭头B']['params'] conditions = [df['签名al_lyB_base'] == 1] if ly_params['require_10x60']: conditions.应用end(df['s_10x60'] == 1) if ly_params['require_ma10_up']: conditions.应用end(df['s_ma10_up'] == 1) if ly_params['require_price_above_20']: conditions.应用end(df['s_price_above_20'] == 1)

df['签名al_lyB'] = np.where(pd.concat(conditions, axis=1).all(axis=1), 1, 0)

# 清理临时列 drop_cols = ['ma5_prev', 'ma10_prev', 'ma20_prev', 'ma60_prev', 'close_prev', 'body', 'body_pct', 'upper_shadow', 'upper_shadow_pct', 's_5x10', 'ma20_slope', 's_20_up', 'price_near_20'] df = df.drop(columns=drop_cols)

return df

============================================================ 选股函数 ============================================================

def