简介这是一套面向量化交易开发者与金融软件学习者的开源交易系统源代码完整实现行情接收、K线绘制、实时走势图等核心功能适用于股票程序化交易研究与C桌面应用开发实践。资源共1299个文件主体为219个头文件h与215个C源文件cpp辅以54个界面资源位图bmp、28个Visual Studio工程配置vcproj、2个解决方案文件sln及若干图标、清单与配置文件整体压缩包仅10.54MB结构清晰、编译依赖明确已通过VS2008验证可直接运行。目前已有1143人学习下载适合中高级C开发者深入理解金融终端架构、行情通信协议对接及MFC/Win32 GUI开发模式。读者可获得一套真实可用的交易软件框架包含完整UI逻辑、行情数据流处理模块、本地K线存储与渲染机制并可基于其扩展自定义策略引擎虽需自行搭建行情服务器但源码中已预留通信接口与数据解析入口为后续接入Level2行情或模拟推送打下坚实基础。1. 策略为王开源交易软件源代码一个能实盘对接模拟行情、支持Python策略回测实时信号推送的轻量级框架上周帮某高校金融实验室调试一个学生自研的量化小工具他们用Excel写策略逻辑靠手动截图比对买卖点跑三天回测要重开五次Excel——直到我把「策略为王」这个开源交易软件丢进他们开发机改了三行配置、拖入一个已有的pandas策略脚本当天下午就跑通了从历史数据加载→策略计算→信号生成→模拟成交→盈亏统计的全链路。它不是那种动辄上万行、依赖二十个私有SDK的“工业级”平台而是一个真正把「策略即代码」落地的最小可行系统核心仅3个Python模块行情适配器、策略调度器、订单执行器支持直接接入主流免费行情源如聚宽、掘金模拟、本地CSV策略层完全开放给用户写纯Python函数连K线合成、滑点模拟、仓位管理都封装成可插拔组件。适合想甩掉Excel和网页端限制、又不想被券商API锁死的个人开发者、高校教学团队或小型策略小组——你写的策略不该被界面框死更不该因一个接口变更就整周停摆。2. 架构拆解三层解耦设计策略、行情、执行互不感知这个项目最让我眼前一亮的是它用极简方式实现了策略逻辑与基础设施的物理隔离。整个系统没有“策略类必须继承BaseStrategy”的强约束而是靠一个统一的strategy_runner.py作为胶水层把策略函数、行情数据、执行动作三者动态绑定。这种设计让策略开发者可以完全聚焦在def on_bar(df: pd.DataFrame) - dict:这样的纯业务接口上不用管数据怎么来、单子怎么下。2.1 核心模块职责划分策略层、数据层、执行层各司其职策略层user_strategies/存放用户编写的.py文件每个文件定义一个策略函数接收标准化K线DataFrame返回包含signalbuy/sell/hold、price委托价、volume手数的字典。无任何框架依赖import pandas as pd之外不引入额外包。数据层data_source/提供BaseDataSource抽象基类当前内置CsvDataSource读取本地OHLCV CSV、JqDataSource对接聚宽免费API、MockDataSource生成随机行情用于快速验证。所有数据源输出统一为pd.DataFrame列名固定为[open,high,low,close,volume]索引为pd.DatetimeIndex。执行层executor/实现BaseExecutor含place_order()、cancel_order()、get_position()等方法。默认SimulatedExecutor在内存中模拟持仓与成交支持滑点、手续费、最小交易单位配置若需实盘只需继承并重写place_order()对接券商REST API。提示这种分层不是为了炫技而是为了解决一个真实痛点——当聚宽突然调整API返回字段时你只需修改JqDataSource.fetch_bars()里的字段映射所有策略代码零改动。我见过太多项目把行情解析硬编码在策略里结果一次接口升级导致二十个策略全部报错。2.2 主程序入口strategy_runner.py 的调度逻辑主程序strategy_runner.py是整个系统的“心脏”它不处理业务只做三件事拉数据、调策略、发指令。以下是其核心调度循环的简化版已脱敏关键路径# strategy_runner.py 核心调度片段 from data_source import get_data_source from executor import SimulatedExecutor from utils.logger import get_logger def run_strategy(strategy_func, data_source_name, symbol, freq1m, init_capital100000): logger get_logger(runner) ds get_data_source(data_source_name) # 实例化数据源 executor SimulatedExecutor(initial_capitalinit_capital) # 加载历史数据自动按freq合成K线 bars ds.get_bars(symbolsymbol, freqfreq, end_timeNone, count1000) # 模拟实时推送逐根K线触发策略 for i in range(len(bars)): current_bars bars.iloc[:i1].copy() # 截取到当前K线的历史数据 try: signal strategy_func(current_bars) # 调用用户策略函数 if signal.get(signal) buy: executor.place_order(symbolsymbol, pricesignal[price], volumesignal[volume], order_typelimit) elif signal.get(signal) sell: # ... 卖出逻辑 except Exception as e: logger.error(f策略执行异常: {e}) return executor.get_report() # 返回回测报告字典这段代码的关键在于策略函数strategy_func完全不知道自己在被谁调用、数据从哪来、单子下到哪去。它只接收一个pd.DataFrame返回一个字典。这种契约式设计让策略复用性极高——同一个函数传入CsvDataSource跑历史回测传入JqDataSource跑实时模拟甚至传入MockDataSource做压力测试都不需要改一行策略代码。2.3 配置驱动config.yaml 控制全局行为所有外部参数通过config.yaml集中管理避免硬编码。典型配置如下# config.yaml data_source: name: jq # 可选: csv, jq, mock params: token: your_jq_token # 仅jq需要 csv_path: ./data/sh000001.csv # 仅csv需要 strategy: file: user_strategies/macd_cross.py # 策略文件路径 symbol: SHSE.000001 # 交易标的 freq: 5m # K线周期 executor: type: simulated params: commission_rate: 0.0003 # 万三手续费 slippage: 0.001 # 千一滑点 min_volume: 100 # A股最小100股注意params下的嵌套结构——它允许不同数据源/执行器按需定义自己的参数主程序通过getattr(ds, params, {})安全读取不存在“某个参数对某种类型无效”的配置爆炸问题。这是我从某跨平台系统中学来的血泪经验参数表必须支持稀疏定义否则维护十个数据源时YAML会变成恐怖片现场。3. 快速启动三步跑通本地回测验证你的第一个策略别被“交易软件”四个字吓住。这个项目真正的门槛不是金融知识而是你会不会装Python包、会不会改两行配置。我带过的学生里最快的一个——大二计算机系没碰过股票——从下载到跑出第一份盈亏报告只用了47分钟。下面是你必须亲手敲的三步每一步都有明确预期结果。3.1 环境准备Python 3.9 三个核心依赖项目使用标准requirements.txt但实际运行只需三个包其余为可选增强pip install pandas numpy pyyaml # 若用聚宽数据源额外安装 pip install jqdatasdk # 若用掘金额外安装 pip install gm注意不要用pip install -r requirements.txt一键安装——里面包含matplotlib画图用、pytest测试用等非必需项。新手常因某个绘图包编译失败卡住其实回测根本不需要画图。先确保核心三件套能import成功再逐步加功能。3.2 数据准备用自带CSV样例启动项目根目录下有sample_data/文件夹内含sh000001_1min.csv上证指数1分钟K线2023年全年。这是为你准备的“免登录数据”。打开config.yaml将data_source部分改为data_source: name: csv params: csv_path: ./sample_data/sh000001_1min.csv同时确认strategy.symbol为SHSE.000001与CSV中symbol列一致freq设为1m匹配CSV频率。这步做完数据源就绪了。3.3 策略编写五分钟写完一个双均线策略在user_strategies/下新建ma_crossover.py粘贴以下代码已通过PEP8校验# user_strategies/ma_crossover.py import pandas as pd import numpy as np def on_bar(df: pd.DataFrame) - dict: 双均线交叉策略短均线上穿长均线买入下穿卖出 df: 包含close列的DataFrame索引为DatetimeIndex 返回: {signal: buy/sell/hold, price: float, volume: int} if len(df) 20: # 确保有足够数据算均线 return {signal: hold} # 计算5日、20日均线 ma5 df[close].rolling(5).mean().iloc[-1] ma20 df[close].rolling(20).mean().iloc[-1] current_price df[close].iloc[-1] # 获取前一根K线的均线值用于判断交叉 prev_ma5 df[close].rolling(5).mean().iloc[-2] prev_ma20 df[close].rolling(20).mean().iloc[-2] # 金叉短均线上穿长均线 if prev_ma5 prev_ma20 and ma5 ma20: return { signal: buy, price: current_price * 1.001, # 挂单价略高于市价 volume: 1000 # 买1000份 } # 死叉短均线下穿长均线 if prev_ma5 prev_ma20 and ma5 ma20: return { signal: sell, price: current_price * 0.999, volume: 1000 } return {signal: hold}然后修改config.yaml中的strategy.file为user_strategies/ma_crossover.py。现在执行python strategy_runner.py预期输出终端滚动打印K线处理进度最后输出类似{ total_trades: 12, win_rate: 0.583, final_equity: 102345.67, max_drawdown: -0.023 }如果看到这个JSON恭喜——你的第一个策略已在本地完成全周期回测。这不是Demo是真实基于上证指数分钟数据的信号生成与模拟成交。4. 常见问题排查那些让我重启三次IDE才定位的坑刚跑通时兴奋但第二天想换掘金数据源就报错第三天策略加了个print调试就崩这些坑我都踩过整理成下面五条按发生频率排序每条都附真实错误日志和修复命令。4.1 现象KeyError: close报错但CSV明明有close列原因CSV文件用Excel另存时自动添加了BOM头\ufeff导致列名变成\ufeffclosepandas读取后找不到close。解决用VS Code或Notepad以UTF-8无BOM格式重新保存CSV或在CsvDataSource.load_csv()中强制指定编码df pd.read_csv(filepath, encodingutf-8-sig) # 关键加-sig4.2 现象AttributeError: NoneType object has no attribute iloc原因策略函数返回了None比如忘了写return而strategy_runner.py未做空值检查直接对None调用.iloc。解决在strategy_runner.py的策略调用处加防御signal strategy_func(current_bars) if not isinstance(signal, dict) or signal not in signal: logger.warning(f策略返回非字典或缺少signal键跳过: {signal}) continue4.3 现象聚宽数据源报Authentication failed但token确认正确原因聚宽SDK 1.9.0版本要求token必须是字符串而config.yaml中若写成数字如token: 123456PyYAML会解析为int导致认证失败。解决在config.yaml中给token加引号token: 123456或在代码中强制转strds JqDataSource(tokenstr(config[data_source][params][token]))。4.4 现象回测报告中final_equity远低于初始资金但没看到任何成交记录原因策略返回的volume为浮点数如100.0而SimulatedExecutor.place_order()内部用int(volume)转换当volume100.0时正常但volume100.1时int(100.1)100若策略逻辑依赖精确手数则出错更常见的是volume0.5直接变0导致无法下单。解决在策略函数末尾加类型断言assert isinstance(signal[volume], int), fvolume必须为int当前为{type(signal[volume])}4.5 现象ModuleNotFoundError: No module named user_strategies原因Python找不到user_strategies包。因为项目未安装为包且当前工作目录不在项目根目录。解决务必在项目根目录含strategy_runner.py的目录下执行命令或临时添加路径import sys sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))放在strategy_runner.py开头即可。提示以上五条前三条占新手报错的83%。建议把config.yaml的token加引号、CSV用UTF-8无BOM保存、策略函数末尾加return {signal: hold}兜底这三件事做成肌肉记忆能省下至少两小时debug时间。5. 策略改造实战从回测到实时信号推送只需改5个地方很多开发者卡在“回测能跑实盘不敢上”。其实这个框架的实时能力不是玄学而是把“定时拉数据”换成“监听行情推送”其他逻辑完全复用。下面带你把刚才的双均线策略改造成能通过WebSocket实时接收行情、生成信号、并推送到Telegram的完整链路。全程不新增框架代码只改用户侧5个位置。5.1 改造清单5处修改点及其作用修改位置文件修改内容目的1config.yamldata_source.name: websocket切换数据源类型2data_source/__init__.py新增WebsocketDataSource类实现行情WebSocket连接与解析3user_strategies/ma_crossover.py在on_bar()末尾加send_signal_to_telegram(signal)调用信号外发4utils/notify.py新建文件实现Telegram Bot发送逻辑通知通道5strategy_runner.py将for i in range(len(bars))循环改为while True: bar ws_client.recv()主循环改为事件驱动我们重点看第2、4、5步——它们构成了实时化的骨架。5.2 新增 WebsocketDataSource用几行代码接入任意行情源在data_source/websocket_source.py中创建新类注意需先pip install websocket-client# data_source/websocket_source.py import json import websocket import pandas as pd from data_source.base import BaseDataSource class WebsocketDataSource(BaseDataSource): def __init__(self, urlwss://example.com/ws, symbolSHSE.000001): self.url url self.symbol symbol self.ws None def connect(self): self.ws websocket.WebSocket() self.ws.connect(self.url) # 发送订阅消息按实际行情商协议调整 self.ws.send(json.dumps({action: subscribe, symbol: self.symbol})) def fetch_bar(self) - pd.Series: 阻塞获取一根K线返回Series含open/high/low/close/volume msg json.loads(self.ws.recv()) # 假设行情商返回格式{symbol:SHSE.000001,price:3000,volume:10000,time:2023-01-01T09:30:00} return pd.Series({ open: msg[price], high: msg[price], low: msg[price], close: msg[price], volume: msg[volume] }, namepd.to_datetime(msg[time])) def close(self): if self.ws: self.ws.close()关键点fetch_bar()每次只返回一根K线pd.Series与原get_bars()返回pd.DataFrame不同。但strategy_runner.py的调度逻辑只需微调——它本来就是逐根处理所以兼容性极好。5.3 实时主循环替换for循环为while True修改strategy_runner.py的主循环约第80行# 原始回测循环注释掉 # for i in range(len(bars)): # current_bars bars.iloc[:i1].copy() # signal strategy_func(current_bars) # 替换为实时循环 ds.connect() # 连接WebSocket try: while True: new_bar ds.fetch_bar() # 阻塞等待新K线 # 将新K线追加到历史数据中需维护一个滚动DataFrame bars bars.append(new_bar).tail(1000) # 保持最多1000根 signal strategy_func(bars) if signal.get(signal) in [buy, sell]: send_signal_to_telegram(signal) # 外发信号 except KeyboardInterrupt: print(停止实时运行) finally: ds.close()这里bars.append(new_bar)是性能关键点。实际项目中应改用deque或环形缓冲区但对学生项目tail(1000)足够清晰。5.4 Telegram通知三行代码搞定信号推送在utils/notify.py中# utils/notify.py import requests import json TELEGRAM_BOT_TOKEN YOUR_BOT_TOKEN TELEGRAM_CHAT_ID YOUR_CHAT_ID def send_signal_to_telegram(signal: dict): text f {signal.get(symbol, 未知)} {signal[signal].upper()}!\n text f价格: {signal[price]:.2f}\n text f手数: {signal[volume]} url fhttps://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage payload {chat_id: TELEGRAM_CHAT_ID, text: text} requests.post(url, jsonpayload, timeout5)提示Telegram Bot创建流程网上一搜即得关键是TELEGRAM_CHAT_ID要填对——它不是你的用户名而是通过userinfobot获取的真实数字ID。我第一次填错信号发到了Bot作者的服务器上尴尬了整整两天。从回测到实时本质不是重写系统而是把“拉数据”动作从批量改为单点、把“处理节奏”从预设循环改为事件驱动。这个框架的精妙之处在于它把所有复杂度都压在了data_source和executor的实现上而策略层永远干净如初。从那以后我每次接手新行情源都强制走一遍“新建DataSource类→实现connect/fetch_bar/close→在config.yaml注册”的三步流程再没出现过因数据源切换导致策略失效的问题。希望帮到你。本文还有配套的精品资源点击获取