"""
双策略选股器 - 通用版
策略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文件必须包含以下列:
"""
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):
"""
加载用户数据
参数:
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'])
# 预计算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):
"""选股主函数"""
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)')
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()
共 1 个版本