我们首先将问题简化,聚焦于交易策略逻辑化,特别是选股过程、入场信号和出场信号的程序化实现。由于您有编程基础,我们可以分步骤进行,并且尽量利用现有的库和工具来降低难度。
步骤1:策略逻辑的明确化
步骤2:数据获取
步骤3:计算技术指标
步骤4:生成交易信号(入场、出场)
步骤5:验证信号逻辑(初步回测)
下面我们详细说明每个步骤需要做的准备工作。
一、策略逻辑的明确化(文档化)
这是最关键的一步,需要将您15年的技术分析经验转化为明确的规则。例如:
入场信号:当某股票满足条件A、B、C时买入。
出场信号:当条件D或条件E出现时卖出。
建议:
-
将您的交易策略用文字描述清楚,最好能列出所有条件。
-
对于每个条件,尽可能用数学或逻辑表达式描述,避免模糊。
-
考虑使用流程图来表示整个决策过程。
二、数据获取
为了计算技术指标和生成信号,我们需要历史行情数据。您可以选择以下方式之一:
-
免费数据源(适合初期验证):
-
Tushare(需要注册获取token,有日线数据)
-
Baostock(无需注册,日线数据)
-
AKShare(需要安装,数据源较多)
-
-
付费数据源(更准确、更全面):
-
万得(Wind)
-
聚宽(JoinQuant)
-
米筐(RiceQuant)
-
准备工作:
-
选择数据源,并学习其API的使用。
-
获取您策略所需的数据(例如:日K线数据,包括开盘价、最高价、最低价、收盘价、成交量等)。
-
数据存储:可以选择本地存储(如CSV文件、SQLite数据库)或直接使用API实时获取。
三、计算技术指标
根据您的策略,可能需要计算各种技术指标,例如:
移动平均线(MA)、相对强弱指标(RSI)、布林带(Bollinger Bands)等。
准备工作:
-
确定策略中需要哪些指标。
-
学习使用TA-Lib库(技术指标库)来计算指标,或者自己编写计算函数。
-
注意指标计算的周期参数(如20日均线)。
四、生成交易信号
根据计算出的指标,按照策略逻辑生成入场和出场信号。
准备工作:
-
编写代码,将策略逻辑转化为条件判断。
-
注意:信号分为入场信号和出场信号,可能需要分别处理。
-
考虑仓位管理(例如,每次买入固定数量或固定金额)。
五、验证信号逻辑(初步回测)
在实盘前,需要验证策略的有效性。可以通过历史数据回测来评估策略的盈亏。
准备工作:
-
编写回测框架(可以自己写,也可以使用现有的框架如Backtrader、Zipline等)。
-
注意回测中的常见问题,如未来函数、滑点、交易成本等。
-
分析回测结果,包括收益率、最大回撤、夏普比率等。
考虑到您希望简化,我们可以先自己编写一个简单的回测,或者使用一些在线平台(如聚宽、米筐)进行回测,这些平台提供了数据、指标计算和回测引擎,可以快速验证想法。
具体准备工作列表:
-
硬件和软件准备:
-
一台可以运行Python的电脑。
-
安装Python(推荐3.7以上版本)。
-
安装必要的库:pandas, numpy, matplotlib, tushare(或其他数据源库), TA-Lib(可能需要单独安装,注意安装方法)等。
-
-
学习准备:
-
复习Python基础,特别是pandas库的使用(因为数据处理常用DataFrame)。
-
学习所选数据源的使用方法。
-
学习TA-Lib库的使用,或者自己实现指标计算。
-
-
策略文档准备:
-
详细描述策略,包括:
a. 选股条件(例如:市值、股价、成交量等)
b. 入场条件(例如:金叉、突破等)
c. 出场条件(例如:死叉、止损点等)
d. 仓位管理规则
-
-
开发计划:
-
第一阶段:实现数据获取和指标计算。
-
第二阶段:实现信号生成。
-
第三阶段:实现简单回测,输出交易记录和绩效。
-
-
测试计划:
-
用少量股票(如沪深300成分股)测试策略。
-
用历史数据验证信号是否符合预期。
-
逐步扩大测试范围。
-
难点:
-
数据质量:免费数据可能存在错误,需要清洗。
-
未来函数:避免在回测中使用未来数据。
-
交易成本:需要考虑佣金和印花税。
-
滑点:实际交易中订单成交价格可能与预期有偏差。
付费内容:
-
数据源:如果免费数据不能满足需求,可能需要购买付费数据。
-
回测平台:一些在线平台提供更便捷的回测和实盘交易功能,但可能需要付费。
-
服务器:如果需要实时运行策略,可能需要购买云服务器。
由于您已经决定先实现选股和信号部分,我们可以先从最简单的本地回测开始,逐步完善。
让我们开始第一步:策略文档化。
如果您已经准备好策略文档,我们可以开始设计代码结构。如果没有,请先整理您的策略。
假设您已经有一个简单的策略,例如:当5日均线上穿20日均线时买入,当5日均线下穿20日均线时卖出。
我们可以用代码实现这个策略作为示例。
很好!从选股和信号生成切入是明智的起步点。这既保持了您的核心竞争力,又避免了系统复杂度。让我帮您规划一个精简高效的实现路径。
一、核心准备工作清单(按优先级)
1. 策略规则文档化(最核心)
这是程序化的基础,必须明确且无歧义:
【示例模板】 选股条件: 1. 技术指标要求: - MA5上穿MA20(金叉) - RSI(14) > 50 且 < 70 - 成交量 > 5日均量 * 1.2 2. 形态识别(需要具体定义): - "W底"形态:必须明确定义(如:两个低点,中间反弹10%以上) - "突破"形态:突破什么?平台/压力线?突破多少?(如:突破20日高点且涨幅>3%) 3. 过滤条件: - 市值范围:50-500亿 - 排除ST/*ST - 剔除停牌股 - 价格>5元(避免低价股) 入场信号(精确到代码可执行): - 条件1 AND 条件2 AND 条件3 - 买入时机:满足条件的**次日开盘**?还是**当日盘中**? - 仓位比例:每次投入资金的百分比? 出场信号(同样要精确): - 止损:跌破买入价的-8%? - 止盈:上涨20%? - 技术出场:死叉/RSI超买/跌破支撑
关键建议:先完成1-2个您最常用、最有效的策略规则文档。
2. 技术环境准备(1天内可完成)
必备软件清单: 1. Python 3.9+(稳定版) 2. VS Code(免费,编程友好) 3. Anaconda(包管理方便) 4. 数据源:Tushare Pro(注册免费,基础数据足够)
安装步骤简化:
# 1. 安装Anaconda(官网下载) # 2. 打开Anaconda Prompt conda create -n quant python=3.9 conda activate quant pip install pandas numpy matplotlib tushare ta-lib
3. 数据获取准备
# 最简单数据获取模板 import tushare as ts pro = ts.pro_api('你的token') # 免费注册获取 # 获取日线数据 df = pro.daily(ts_code='000001.SZ', start_date='20230101') print(df.head())
初期只需准备:
-
股票列表(沪深A股)
-
日线行情数据(开盘、最高、最低、收盘、成交量)
-
停牌/复牌信息(Tushare免费提供)
4. 指标计算准备
TA-Lib库覆盖了99%的技术指标:
import talib # 示例:计算MACD df['macd'], df['signal'], df['hist'] = talib.MACD(df['close'])
二、精简实现步骤(2-4周完成基础版)
第1周:环境搭建 + 单一策略实现
目标:实现您最熟悉的一个选股策略 步骤: 1. 下载安装软件(2小时) 2. 获取股票列表(5000+股票) 3. 循环每只股票,计算指标 4. 应用选股条件筛选 5. 输出符合条件的股票列表 成果:每天能输出符合条件的股票(如:金叉股列表)
第2周:信号验证 + 回测
目标:验证信号的有效性 步骤: 1. 历史回测:信号出现后N天的表现 2. 计算简单指标:胜率、盈亏比 3. 可视化展示:信号点标注在K线图 成果:知道这个策略历史表现如何
第3周:优化 + 多策略集成
目标:增加更多策略条件 步骤: 1. 加入更多过滤条件 2. 优化参数(如:不同周期均线) 3. 多策略组合(OR/AND关系) 成果:更稳定的选股系统
三、具体实现示例(一个完整的最小可行产品)
以下是一个可直接运行的选股程序框架:
""" 简易选股系统 v1.0 功能:每日筛选符合条件的股票 """ import pandas as pd import tushare as ts import talib from datetime import datetime, timedelta class StockSelector: def __init__(self, token): """初始化""" self.pro = ts.pro_api(token) self.today = datetime.now().strftime('%Y%m%d') def get_all_stocks(self): """获取全A股列表""" # 获取基础信息 df = self.pro.stock_basic(exchange='', list_status='L') return df[['ts_code', 'name', 'list_date']] def get_daily_data(self, ts_code, days=100): """获取股票日线数据""" end_date = self.today start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d') df = self.pro.daily(ts_code=ts_code, start_date=start_date, end_date=end_date) df = df.sort_values('trade_date') return df def calculate_indicators(self, df): """计算技术指标""" if len(df) < 30: # 数据太少 return df # 计算均线 df['ma5'] = talib.SMA(df['close'], timeperiod=5) df['ma20'] = talib.SMA(df['close'], timeperiod=20) df['ma60'] = talib.SMA(df['close'], timeperiod=60) # 计算MACD df['macd'], df['macd_signal'], df['macd_hist'] = talib.MACD( df['close'], fastperiod=12, slowperiod=26, signalperiod=9 ) # 计算RSI df['rsi'] = talib.RSI(df['close'], timeperiod=14) # 计算成交量均线 df['vol_ma5'] = talib.SMA(df['vol'], timeperiod=5) return df def check_buy_signal(self, df): """检查买入信号(示例:金叉策略)""" if len(df) < 2: return False latest = df.iloc[-1] # 最新数据 prev = df.iloc[-2] # 前一天数据 # 条件1:5日均线上穿20日均线(金叉) condition1 = (prev['ma5'] <= prev['ma20']) and (latest['ma5'] > latest['ma20']) # 条件2:价格在60日均线之上 condition2 = latest['close'] > latest['ma60'] # 条件3:RSI在合理区间 condition3 = 30 < latest['rsi'] < 70 # 条件4:放量(成交量大于5日均量1.2倍) condition4 = latest['vol'] > latest['vol_ma5'] * 1.2 return condition1 and condition2 and condition3 and condition4 def run_screening(self): """运行选股""" print(f"开始选股...时间:{self.today}") # 1. 获取所有股票 stocks = self.get_all_stocks() print(f"共有 {len(stocks)} 只股票") # 2. 筛选结果存储 selected_stocks = [] # 3. 遍历股票(为了测试,先只遍历前100只) for idx, row in stocks.head(100).iterrows(): try: # 获取数据 df = self.get_daily_data(row['ts_code'], days=100) if len(df) < 30: # 数据不足 continue # 计算指标 df = self.calculate_indicators(df) # 检查信号 if self.check_buy_signal(df): latest = df.iloc[-1] selected_stocks.append({ '代码': row['ts_code'], '名称': row['name'], '收盘价': latest['close'], 'MA5': latest['ma5'], 'MA20': latest['ma20'], 'RSI': round(latest['rsi'], 2) }) print(f"发现信号:{row['ts_code']} {row['name']}") except Exception as e: # 出错继续下一只 continue # 4. 输出结果 if selected_stocks: result_df = pd.DataFrame(selected_stocks) print(f"\n选股完成!共发现 {len(selected_stocks)} 只符合条件的股票:") print(result_df.to_string(index=False)) # 保存到Excel result_df.to_excel(f'selected_stocks_{self.today}.xlsx', index=False) print(f"结果已保存到 selected_stocks_{self.today}.xlsx") else: print("今日无符合条件的股票") return selected_stocks # 使用示例 if __name__ == "__main__": # 替换为你的tushare token(免费注册获取) TOKEN = "你的tushare_token" selector = StockSelector(TOKEN) selector.run_screening()
四、重点注意事项
1. 策略规则必须明确
-
避免模糊描述:”趋势向好” → 具体化为”价格在60日均线上方”
-
避免主观判断:”感觉要突破” → 具体化为”突破20日高点且涨幅>5%”
2. 数据质量问题
-
复权处理:使用前复权数据
-
停牌处理:跳过停牌期间
-
新股处理:上市不足60天的排除
3. 避免未来函数
-
不能用未来数据:比如用当天收盘价计算信号,实际应该在收盘前判断
-
解决方法:使用
df.shift(1)获取前一天的数据做判断
五、快速启动计划(2周版)
第1天:
-
注册Tushare Pro(免费)获取token
-
安装Python环境
-
运行上面的示例代码,看到结果
第2-3天:
-
修改
check_buy_signal函数,加入您的第一条策略 -
测试并调整
第4-7天:
-
逐步添加更多策略条件
-
增加回测验证功能
第2周:
-
优化代码结构
-
增加可视化(K线图标注信号点)
-
开始思考出场策略的实现
六、可能遇到的难点及解决方案
难点1:形态识别程序化
-
解决方案:先实现简单的形态(金叉死叉、突破),复杂形态(头肩顶、W底)后期再实现
难点2:性能问题(遍历5000只股票)
-
解决方案:先小范围测试,后期用多线程加速
难点3:数据更新
-
解决方案:每天收盘后自动运行一次,生成选股列表