双策略A股选股器

基于仙人指路B和老鸭头B两策略，筛选符合多条件的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) """

import pandas as pd import numpy as np from datetime import datetime import argparse import 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, target_date=None): """ 加载用户数据

text

参数:
    data_path: CSV文件路径
    target_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("[INFO] 请确保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 detect_signals(df): """检测双策略信号""" df = df.sort_values(['symbol', 'date'])

text

# 预计算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['signal_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'].transform(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['signal_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['signal_lyB_base'] == 1]
if ly_params['require_10x60']:
    conditions.append(df['s_10x60'] == 1)
if ly_params['require_ma10_up']:
    conditions.append(df['s_ma10_up'] == 1)
if ly_params['require_price_above_20']:
    conditions.append(df['s_price_above_20'] == 1)

df['signal_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 select_stocks(df, target_date=None, top_n=20): """选股主函数"""

text

if target_date:
    target_dt = pd.to_datetime(target_date)
else:
    target_dt = df['date'].max()

print(f"[OK] 目标日期: {target_dt.strftime('%Y-%m-%d')}")

# 检测信号
df = detect_signals(df)

# 只取目标日期
today_df = df[df['date'] == target_dt].copy()

if len(today_df) == 0:
    print(f"[WARN] 目标日期无数据")
    return None

# === 策略1: 仙人指路B ===
xr_signals = today_df[today_df['signal_xrB'] == 1].copy()

# === 策略2: 老鸭头B ===
ly_signals = today_df[today_df['signal_lyB'] == 1].copy()

# 输出结果
print("\n" + "="*80)
print(f"双策略选股结果 - {target_dt.strftime('%Y-%m-%d')}")
print("="*80)

# 策略1输出
print(f"\n【策略1: 仙人指路B】")
print(f"预期胜率: {STRATEGY_CONFIG['仙人指路B']['expected_win_rate']}%")
print(f"预期均益: +{STRATEGY_CONFIG['仙人指路B']['expected_avg_profit']}%")
print(f"持股周期: {STRATEGY_CONFIG['仙人指路B']['hold_days']}天")
print(f"信号股票: {len(xr_signals)}只")

if len(xr_signals) > 0:
    xr_signals = xr_signals.sort_values('volume_ratio', ascending=False)
    top_xr = xr_signals.head(top_n)
    
    print(f"\n--- TOP {len(top_xr)} ---")
    for i, (_, row) in enumerate(top_xr.iterrows(), 1):
        print(f"\n{i:2d}. {row['symbol']} {row['name']}")
        print(f"    价格: {row['close']:.2f} | 涨幅: {row['pct_change']:+.2f}%")
        print(f"    换手: {row['turnover']:.2f}% | 量比: {row['volume_ratio']:.2f}")
        print(f"    上影线: {(row['high'] - max(row['open'], row['close'])) / row['close'] * 100:.2f}%")
        print(f"    实体: {abs(row['close'] - row['open']) / row['open'] * 100:.2f}%")

# 策略2输出
print(f"\n【策略2: 老鸭头B】")
print(f"预期胜率: {STRATEGY_CONFIG['老鸭头B']['expected_win_rate']}%")
print(f"预期均益: +{STRATEGY_CONFIG['老鸭头B']['expected_avg_profit']}%")
print(f"持股周期: {STRATEGY_CONFIG['老鸭头B']['hold_days']}天")
print(f"信号股票: {len(ly_signals)}只")

if len(ly_signals) > 0:
    ly_signals['score'] = (
        ly_signals['s_10x60'] * 30 +
        ly_signals['s_ma10_up'] * 20 +
        ly_signals['turnover'] * 5
    )
    ly_signals = ly_signals.sort_values('score', ascending=False)
    top_ly = ly_signals.head(top_n)
    
    print(f"\n--- TOP {len(top_ly)} ---")
    for i, (_, row) in enumerate(top_ly.iterrows(), 1):
        print(f"\n{i:2d}. {row['symbol']} {row['name']}")
        print(f"    价格: {row['close']:.2f} | 涨幅: {row['pct_change']:+.2f}%")
        print(f"    换手: {row['turnover']:.2f}% | 量比: {row['volume_ratio']:.2f}")
        print(f"    MA5: {row['ma5']:.2f} | MA10: {row['ma10']:.2f} | MA20: {row['ma20']:.2f} | MA60: {row['ma60']:.2f}")
        print(f"    信号: 10日上穿60日{'Y' if row['s_10x60'] else 'N'} | MA10向上{'Y' if row['s_ma10_up'] else 'N'}")
        print(f"    评分: {row['score']:.0f}")

# 保存结果
if len(xr_signals) > 0 or len(ly_signals) > 0:
    output_file = f"selection_{target_dt.strftime('%Y%m%d')}.csv"
    
    # 合并保存
    xr_save = xr_signals[['symbol', 'name', 'date', 'close', 'pct_change', 'turnover', 'volume_ratio']].copy()
    xr_save['strategy'] = '仙人指路B'
    
    ly_save = ly_signals[['symbol', 'name', 'date', 'close', 'pct_change', 'turnover', 'volume_ratio']].copy()
    ly_save['strategy'] = '老鸭头B'
    
    combined = pd.concat([xr_save, ly_save], ignore_index=True)
    combined.to_csv(output_file, index=False)
    print(f"\n[OK] 结果保存: {output_file}")

return xr_signals, ly_signals

============================================================

主函数

============================================================

def main(): parser = argparse.ArgumentParser(description='双策略选股器 - 通用版') parser.add_argument('--data', type=str, required=True, help='数据文件路径 (CSV)') parser.add_argument('--date', type=str, help='指定日期 (YYYY-MM-DD)') parser.add_argument('--top', type=int, default=20, help='显示前N只 (默认20)')

text

args = parser.parse_args()

# 加载数据
df = load_data(args.data, args.date)
if df is None:
    return

# 选股
select_stocks(df, args.date, args.top)

if name == 'main': main()