从“看K线”到“练K线手”为什么你缺的不是行情数据而是反馈环境如果你看过通达信自带的“K线训练”功能应该会有一种很矛盾的感觉软件随机抽出一段历史K线遮住后面走势让你判断“会涨还是会跌”。听起来很像游戏但你实际去练一练就会发现它更像一套枯燥的刷题工具——没有对手、没有仓位、没有盈亏反馈更没有一个根据你水平动态调整难度的AI教练。这也是很多人在学习技术分析时的真实困境看完一堆“底部形态”“突破买入”的教程打开真实行情依然不知道怎么操作。因为图形识别是一种被动能力而交易决策是一种主动能力。被动能力靠阅读获得主动能力必须靠有反馈的练习来打磨。所以当我决定做B站AI创造公开赛参赛作品时最想做的不是“又一个预测涨跌的模型”而是一个能让人在“玩”的过程中真正练出手感的交易游戏。这就是《K线猎手》的由来。在这篇文章里我会把整个项目拆开讲清楚如何用Python生成仿真K线如何用深度强化学习训练一个AI对手以及如何把AI放进一场“人和机器比交易”的对抗游戏里。如果你正准备学习强化学习或者想用AI做一个能跑通、能演示、能放进作品集的项目这篇文章会是一份比较完整的动手参考。1. 从“K线训练”到“K线猎手”这个游戏要解决什么问题先回答一个很多人问过的问题通达信的K线训练里出现的那些K线是真实发生的吗从产品机制上说它抽取的是真实历史行情片段这些片段确实发生过。但关键是它只是把一段已经完成的行情完整展示给你然后让你做“下一步看涨还是看跌”的选择。真实交易中真正困难的不是“看图说话”而是仓位管理、止损止盈、资金曲线波动带来的心理压力以及连续决策中产生的反馈循环。传统K线训练缺少三样东西缺少对抗性。你可以判断行情但没有任何一个角色在看你的判断。缺少连续性。单根K线判断是孤立动作无法体现“多次决策后的综合收益”。缺少难度自适应。新手和老手面对的是同一批随机片段训练效率很低。《K线猎手》想解决的就是这三点。它不是拿真实K线让你刷题而是构建一个“行情引擎 AI选手 玩家”三方互动的环境行情引擎动态生成仿真K线保证每次对局都不一样AI选手通过深度强化学习不断调整自己的交易策略玩家在一个类似终端的界面上与AI在同一段行情中做交易最后对比累计收益。这让我想到一个更好的类比把K线训练从“做选择题”变成“玩游戏”。选择题有标准答案游戏没有标准答案只有因为在某个时间点做了某个动作而带来的结果反推。所以我们不能只把K线数据丢给玩家看一眼而是需要一整套包含环境、状态、动作、奖励的机制。这也是强化学习天然适合这个场景的原因。2. 核心设计思路AI生成行情AI做对手盘整个游戏可以拆成四个模块。模块职责技术要点行情引擎生成仿真K线数据几何布朗运动、波动率聚合交易环境统一处理玩家和AI的动作Gym风格环境状态、动作、奖励AI选手学习交易策略的智能体Dueling DQN经验回放目标网络对战界面让玩家和AI在同一段行情上PK命令行或网页交互我们在设计时特别注意一个原则不要试图让AI“预测未来”而是让AI“学会在一个不完全确定的行情里管理仓位和风险”。这两者区别很大。预测模型的目标函数是“下一根K线涨还是跌”它的优化方向是准确率。而交易智能体的目标函数是“在一个完整回合结束后的总财富”它的优化方向是最终收益。即使预测准确率只有52%一套好的仓位管理策略也能盈利反过来哪怕预测准确率80%一次爆仓也会归零。所以《K线猎手》里的AI不是一个“行情预测器”而是一个“决策优化器”。它观察过去一截K线决定自己当前应该空仓等待、买入持有还是卖出离场然后用累计收益来评判自己的每个决策。这个设计也使得游戏对玩家是公平的。AI看到的数据窗口玩家也能看到AI必须面对交易成本玩家同样要面对AI会在连续亏损时被迫减仓玩家也会。双方在同一个规则下竞技比的不是谁更会“猜”而是谁更能在不确定性中做对决策。3. 环境准备与基础配置《K线猎手》的核心代码依赖Python生态建议使用Python 3.10及以上版本。项目本身不依赖大型数据库也不需要GPU普通开发机就能完成训练和演示。建议先创建一个独立虚拟环境避免依赖冲突python -m venv kline_hunter_env source kline_hunter_env/bin/activate # Windows 下使用 kline_hunter_env\Scripts\activate然后安装以下依赖pip install numpy pandas matplotlib torch gym各依赖的作用如下numpy数值计算生成价格序列pandas处理OHLCV数据方便切片和统计matplotlib绘制K线图和收益曲线torch实现深度强化学习网络gym定义标准强化学习环境接口。如果你的网络环境下载PyTorch较慢可以到官网选择对应平台的安装命令。本文核心逻辑不依赖PyTorch的特别版本1.13以上的常见版本都能运行。项目目录结构建议如下kline-hunter/ ├── data_generator.py # 生成仿真K线 ├── env.py # 交易环境 ├── agent.py # DQN智能体 ├── train.py # 训练脚本 ├── play.py # 人机对战入口 └── models/ # 保存训练后的模型权重下面我们按照从数据到对战的顺序逐步实现这个游戏。4. 第一步随机生成仿真K线数据为什么不用真实历史数据直接做游戏真实数据当然更可信但游戏场景里我们需要大量对局。如果把2020年到2024年的数据都喂给玩家和AI很快就会被记住游戏就失去了变化。因此我们选择用几何布朗运动生成仿真价格序列。几何布朗运动是金融领域最经典的随机过程之一它假设价格的百分比变化服从正态分布能产生“看起来像股票走势”的序列。# 文件路径data_generator.py import numpy as np import pandas as pd def generate_kline(num_bars500, start_price100.0, mu0.0002, sigma0.01, dt1.0/252.0): prices np.zeros(num_bars 1) prices[0] start_price for i in range(1, num_bars 1): prices[i] prices[i-1] * np.exp( (mu - 0.5 * sigma**2) * dt sigma * np.sqrt(dt) * np.random.randn() ) opens prices[:-1] closes prices[1:] # 构造 high 和 low在 open/close 的基础上增加随机振幅 high np.maximum(opens, closes) * (1 np.abs(np.random.normal(0, 0.002, num_bars))) low np.minimum(opens, closes) * (1 - np.abs(np.random.normal(0, 0.002, num_bars))) # 保证 high 一定大于等于 open 和 close 的最大值 high np.maximum(high, np.maximum(opens, closes)) low np.minimum(low, np.minimum(opens, closes)) volume np.random.uniform(1000, 5000, num_bars) df pd.DataFrame({ open: opens, high: high, low: low, close: closes, volume: volume }) return df这里真正重要的是np.exp那一行。它使用欧拉离散化实现了几何布朗运动每个新价格等于前一个价格乘以一个小的随机波动。参数mu控制了长期漂移方向sigma控制波动率。当mu为0时走势基本是随机震荡。生成这个文件后可以写一小段代码快速验证from data_generator import generate_kline df generate_kline(num_bars300) print(df.head()) print(df.tail()) print(最高价:, df[high].max(), 最低价:, df[low].min())预期输出是一张300行的DataFrame字段只有open、high、low、close、volume。这个格式和交易软件常见的日K线完全一致后面导入真实数据时也不需要改表结构。5. 第二步把K线行情变成AI可学习的环境强化学习环境的核心是“状态—动作—奖励”三元组。我们需要回答三个问题AI看到什么——状态空间AI能做什么——动作空间AI怎么知道自己做得好不好——奖励函数状态空间我选用最近10根K线的涨跌幅加上当前持仓数量和现金占比。这比直接把原始价格送给网络更稳定因为不同股票的价格区间差异很大归一化涨跌幅能让模型更容易收敛。动作空间设置为3个离散动作0表示空仓等待1表示买入2表示卖出。为了控制复杂度我们规定同一时刻最多只能持有股票买入需要现金卖出需要持仓。奖励函数使用“财富变动值”。每一回合开始时AI有一个初始资金例如10万元每一步结束后AI的现金加上当前持仓市值就是总财富。奖励等于总财富的变动值再减一个极少量的持有惩罚目的是鼓励AI不要一直持仓不动。下面是交易环境的核心实现# 文件路径env.py import gym import numpy as np from gym import spaces from data_generator import generate_kline class KLineTradingEnv(gym.Env): def __init__(self, dfNone, lookback10, initial_cash100000, trade_cost0.0005): super().__init__() if df is None: df generate_kline() self.df df.reset_index(dropTrue) self.lookback lookback self.initial_cash initial_cash self.trade_cost trade_cost self.action_space spaces.Discrete(3) self.observation_space spaces.Box( low-np.inf, highnp.inf, shape(lookback 2,), dtypenp.float32 ) self.reset() def reset(self): self.step_idx self.lookback self.cash self.initial_cash self.shares 0 self.entry_price 0 self.total_wealth self.initial_cash return self._get_state() def _get_state(self): recent_close self.df[close].iloc[ self.step_idx - self.lookback : self.step_idx ].values prev_close self.df[close].iloc[ self.step_idx - self.lookback - 1 ] pct_change recent_close / prev_close - 1 state np.concatenate([ pct_change, [self.shares 0], [self.cash / self.initial_cash] ]) return state.astype(np.float32) def step(self, action): current_price self.df[close].iloc[self.step_idx] done False # 执行动作 if action 1 and self.shares 0: # 买入 shares int(self.cash / current_price / (1 self.trade_cost)) if shares 0: self.cash - shares * current_price self.shares shares self.entry_price current_price elif action 2 and self.shares 0: # 卖出 self.cash self.shares * current_price * (1 - self.trade_cost) self.shares 0 # 计算总财富和奖励 wealth self.cash self.shares * current_price reward wealth - self.total_wealth reward - 1e-6 * wealth # 轻量持有惩罚 self.total_wealth wealth self.step_idx 1 if self.step_idx len(self.df): # 如果最后一根K线仍持仓强制平仓 if self.shares 0: current_price self.df[close].iloc[-1] self.cash self.shares * current_price * (1 - self.trade_cost) self.shares 0 done True return self._get_state(), float(reward), done, {} def current_wealth(self): return self.cash self.shares * self.df[close].iloc[self.step_idx]这段代码里我们加入了交易成本trade_cost默认是万五。这个细节特别重要。如果没有交易成本AI会疯狂买卖因为它只需要跟着涨跌来回操作就能获得正收益加入成本后AI会逐渐学会“减少不必要的交易”这正是真实交易中每个人都要面临的事。状态的形状是(lookback 2,)即10个涨跌幅 1个持仓标记 1个现金占比。如果你想加入更多特征比如参考MACD、RSI、成交量等只需要在_get_state里拼接即可。6. 第三步用深度强化学习训练“AI猎手”环境准备好之后我们开始训练AI。模型选择Dueling DQN。为什么不用最基础的DQNDueling DQN的核心思想是把Q值拆成两部分状态价值V无论做什么动作当前这个局面值多少钱和优势函数A在相同状态下相对平均动作能好多少。在交易场景中大部分时间里最优动作是“什么都不做”基础DQN会平摊很多噪音而Dueling网络能更稳定地学习状态价值。先定义网络结构# 文件路径agent.py import torch import torch.nn as nn import torch.optim as optim import numpy as np import random from collections import deque, namedtuple class DuelingQNet(nn.Module): def __init__(self, state_dim, action_dim, hidden128): super(DuelingQNet, self).__init__() self.feature nn.Sequential( nn.Linear(state_dim, hidden), nn.ReLU(), nn.Linear(hidden, hidden), nn.ReLU() ) self.value nn.Linear(hidden, 1) self.advantage nn.Linear(hidden, action_dim) def forward(self, x): features self.feature(x) value self.value(features) advantage self.advantage(features) return value advantage - advantage.mean(dim1, keepdimTrue)然后定义经验回放缓冲区用于打破训练样本的相关性Transition namedtuple(Transition, [state, action, reward, next_state, done]) class ReplayBuffer: def __init__(self, capacity50000): self.buffer deque(maxlencapacity) def push(self, state, action, reward, next_state, done): self.buffer.append(Transition(state, action, reward, next_state, done)) def sample(self, batch_size): batch random.sample(self.buffer, batch_size) state torch.tensor(np.array([t.state for t in batch]), dtypetorch.float32) action torch.tensor([t.action for t in batch], dtypetorch.long).unsqueeze(1) reward torch.tensor([t.reward for t in batch], dtypetorch.float32).unsqueeze(1) next_state torch.tensor(np.array([t.next_state for t in batch]), dtypetorch.float32) done torch.tensor([t.done for t in batch], dtypetorch.float32).unsqueeze(1) return state, action, reward, next_state, done def __len__(self): return len(self.buffer)接下来是训练循环。这里使用epsilon贪心策略一开始AI会随机尝试各种操作慢慢减少随机探索更多依赖模型输出。# 文件路径train.py from env import KLineTradingEnv from agent import DuelingQNet, ReplayBuffer import torch import torch.nn.functional as F import torch.optim as optim import random import numpy as np def train(episodes300, batch_size64, gamma0.99, lr1e-3): env KLineTradingEnv() state_dim env.observation_space.shape[0] action_dim env.action_space.n policy_net DuelingQNet(state_dim, action_dim) target_net DuelingQNet(state_dim, action_dim) target_net.load_state_dict(policy_net.state_dict()) target_net.eval() optimizer optim.Adam(policy_net.parameters(), lrlr) buffer ReplayBuffer(capacity50000) epsilon 1.0 epsilon_min 0.01 epsilon_decay 0.995 all_rewards [] for ep in range(episodes): state env.reset() total_reward 0 done False while not done: if random.random() epsilon: action env.action_space.sample() else: with torch.no_grad(): q_values policy_net(torch.tensor(state).unsqueeze(0)) action q_values.argmax().item() next_state, reward, done, _ env.step(action) buffer.push(state, action, reward, next_state, done) state next_state total_reward reward if len(buffer) batch_size: batch_state, batch_action, batch_reward, batch_next_state, batch_done buffer.sample(batch_size) q_values policy_net(batch_state).gather(1, batch_action) with torch.no_grad(): next_q_values target_net(batch_next_state).max(1)[0].unsqueeze(1) target_q_values batch_reward gamma * next_q_values * (1 - batch_done) loss F.mse_loss(q_values, target_q_values) optimizer.zero_grad() loss.backward() optimizer.step() epsilon max(epsilon_min, epsilon * epsilon_decay) if ep % 10 0: target_net.load_state_dict(policy_net.state_dict()) all_rewards.append(total_reward) print(fEpisode {ep 1}, Total Reward: {total_reward:.2f}, Epsilon: {epsilon:.3f}) torch.save(policy_net.state_dict(), models/dqn_kline.pt) return all_rewards if __name__ __main__: rewards train(episodes200)训练过程中的几点观察前几十个episode里agent会因为随机探索大量买卖总财富可能波动很大当epsilon降低到0.1以下agent会开始表现出一致性比如不再频繁持仓目标网络每10回合同步一次避免Q值更新目标被自身干扰。这个训练过程本身就是一个绝佳的“实况演示”观众能看到AI从随机乱按到逐渐学会交易的收敛过程。对B站AI创造公开赛的评委来说这种可视化的过程比你贴一串实验数据更有说服力。7. 第四步制作玩家对战模式AI训练完成后我们需要给玩家一个和他对战的入口。为了降低开发门槛第一步做成命令行文本版玩家输入1买入、2卖出、0观望AI根据网络输出自动决策最后对比双方总收益率。# 文件路径play.py import numpy as np import torch from data_generator import generate_kline from agent import DuelingQNet from env import KLineTradingEnv def run_game(model_pathmodels/dqn_kline.pt, num_bars200): df generate_kline(num_barsnum_bars) # AI 状态 policy_net DuelingQNet(12, 3) policy_net.load_state_dict(torch.load(model_path)) policy_net.eval() player_cash 100000 player_shares 0 ai_cash 100000 ai_shares 0 ai_running KLineTradingEnv(dfdf, lookback10, initial_cash100000) state ai_running.reset() print( 欢迎来到《K线猎手》人机对战模式 ) print(按回车进入下一根K线输入 1 买入输入 2 卖出输入 0 观望。) for i in range(10, len(df) - 1): current_price df[close].iloc[i] # 玩家决策 action int(input(f[Bar {i}] 收盘价: {current_price:.2f}请输入动作(0/1/2): ) or 0) # AI决策 with torch.no_grad(): q_values policy_net(torch.tensor(state).unsqueeze(0)) ai_action q_values.argmax().item() # 玩家执行 if action 1 and player_shares 0: player_shares int(player_cash / current_price) player_cash - player_shares * current_price elif action 2 and player_shares 0: player_cash player_shares * current_price player_shares 0 # AI执行 if ai_action 1 and ai_shares 0: ai_shares int(ai_cash / current_price) ai_cash - ai_shares * current_price elif ai_action 2 and ai_shares 0: ai_cash ai_shares * current_price ai_shares 0 # 推进AI环境状态 state, _, _, _ ai_running.step(ai_action) player_wealth player_cash player_shares * current_price ai_wealth ai_cash ai_shares * current_price print(f玩家财富: {player_wealth:.2f} | AI财富: {ai_wealth:.2f} | AI动作: {ai_action}) print(游戏结束) print(f玩家最终: {player_cash player_shares * df[close].iloc[-1]:.2f}) print(fAI最终: {ai_cash ai_shares * df[close].iloc[-1]:.2f}) if __name__ __main__: run_game()这个版本只用了几十行就把“人机对战”串起来了。如果你有前端经验可以再用Streamlit或Vue把K线图画出来同时把交易按钮做成网页交互。但从演示角度来说命令行版本已经能完整表达游戏设计。需要注意的是玩家看到的是逐根更新的K线而不是一上来就能看到整段行情。这里我们在循环中从第10根K线开始也就是玩家已经看到最初10根然后逐步等待下一根出现。这样可以保证“只使用当前已存在的信息”避免未来函数。8. 运行结果与效果验证训练完成后我们需要一套验证标准而不是只看“Loss不断下降”。对交易游戏来说最直接的验证指标是“累计收益曲线”。建议在训练结束后重新生成一段新的K线用训练好的模型跑一个完整回测from env import KLineTradingEnv import torch from agent import DuelingQNet import numpy as np model DuelingQNet(12, 3) model.load_state_dict(torch.load(models/dqn_kline.pt)) model.eval() env KLineTradingEnv() # 内部会生成一段新的随机行情 state env.reset() done False wealth_curve [env.initial_cash] while not done: with torch.no_grad(): q model(torch.tensor(state).unsqueeze(0)) action q.argmax().item() state, reward, done, _ env.step(action) wealth_curve.append(env.current_wealth()) final_return wealth_curve[-1] / env.initial_cash - 1 print(f最终收益率: {final_return * 100:.2f}%)一个训练良好的AI在随机行情中未必能稳定盈利因为行情本身就是随机的。判断训练是否成功的标准有三条AI不再出现“无意义的频繁交易”交易次数明显下降在下跌行情中AI会倾向于空仓等待而不是一路补仓最终收益率的方差变小不会出现“某次大赚、某次爆亏”的极端情况。如果看到AI始终在满仓不动可能是奖励函数里的交易成本设置得太低如果AI一直空仓不敢买可能是买入后的奖励反馈不够充分或者卖出动作没有足够的惩罚。这些都是后续调参的方向。9. 常见问题与排查思路在实际动手过程中最容易遇到下面几个问题问题现象可能原因排查方式解决方案训练时Loss不下降奖励值尺度太大Q值不稳定打印奖励均值和方差对奖励做归一化或裁剪例如限制在 [-1,1] 区间AI永远不买入epsilon衰减太慢或买入奖励太弱检查每个动作的频率调低epsilon_min或增加持仓收益的奖励权重AI频繁买卖交易成本设置太低查看每个episode的交易次数调高trade_cost例如设为0.002生成K线出现负价格几何布朗运动的参数过激检查价格数组尾部降低sigma或使用对数价格构造价格约束Gym版本兼容报错新版本gym移除了部分API查看报错栈使用gym0.21.0或升级到gymnasium并改import输入真实历史数据后无法运行时间顺序或索引问题检查DataFrame是否重置索引使用df.reset_index(dropTrue)并保证按时间升序如果训练过程内存不断增加请检查ReplayBuffer的容量设置。5万的容量在普通机器上可以接受但如果状态维度很大建议降到2万以内。10. 最佳实践与工程建议在把《K线猎手》从“能演示”变成“更专业”的过程中有六个工程细节值得注意。第一永远把数据划分成训练集和评估集。我们不能用训练过的行情段来宣传“效果很好”。更稳妥的做法是先生成一份长期K线数据前70%用于训练后30%用于验证确保模型从未见过验证阶段的价格变化。第二奖励函数不要只写“赚了多少”。真实交易玩家的感受是“回撤比亏损更可怕”。你可以把奖励改成总财富的变化减去一个回撤惩罚项比如current_max_wealth max(self.total_wealth, wealth) drawdown self.total_wealth - wealth reward wealth - self.total_wealth - 0.1 * max(0, drawdown)这样AI会主动控制回撤而不是追求收益最大化。第三AI对战的公平性要透明。玩家第一眼应该能清楚看到AI的初始资金是多少玩家是多少交易成本是多少。不要隐藏规则。一旦规则不透明游戏就失去了训练价值。第四一定要加“免责声明”。这个项目本质是一个学习工具帮助玩家练习交易纪律但不构成投资建议也不能用来预测真实股票涨跌。你可以在页脚或启动命令里输出这样一行print(免责声明本项目仅用于技术学习和娱乐不构成任何投资建议。)第五给玩家提供“难度选择”。可以让AI的水平通过epsilon或模型权重来控制。例如新手模式使用训练50个episode的弱模型困难模式使用训练300个episode的强模型。这比单纯随机行情更有可玩性。第六在B站AI创造公开赛的作品展示中建议录制一段“屏幕交互”的操作过程。因为这类比赛最终看的不是论文而是观众和评委能不能快速理解你的创意。命令行界面虽然简单但录制时最好已经接入网页版可视化界面哪怕是简单的Streamlit。11. 这个项目还能怎么升级《K线猎手》目前是一个完整的“AI交易游戏”最小闭环生成行情、定义环境、训练AI、人机对战。但它的扩展空间非常大。如果你想让行情更真实可以接入Tushare或AkShare等免费数据源把沪深300的成分股日线数据下载下来切成片段后随机抽取。但要注意导入真实数据后就不能再使用随机游走逻辑需要重新对齐索引和时间。如果你想用当下最热的大模型AI能力可以在AI决策中加入LLM辅助。例如每隔10根K线让大模型用自然语言总结当前走势“当前处于均线粘合后的突破放量阶段建议等待回调后买入”再把这个文本作为额外特征交给决策模型。虽然目前多模态特征融合比较复杂但作为展示项目LLM的文字解说已经能极大提升观感。如果你想让AI更像一个有“性格”的对手可以引入多Agent博弈一个激进型AI、一个防守型AI、一个随机型AI让玩家选择挑战对象。这样游戏的深度和娱乐性都会增加。如果你想深入强化学习本身还可以探索PPO、A2C、SAC等算法并对比不同算法在交易环境中的收敛速度和稳定性。交易环境是一个部分可观测环境非常适合用来理解强化学习算法之间的差异。回归到最初的问题通达信的K线训练是真实发生的K线但它只是静态题目。而《K线猎手》把K线变成了一场有对手、有反馈、有奖惩的游戏——这才是AI时代里学习交易决策更有趣的方式。整个项目没有依赖昂贵的商业平台只用Python、PyTorch和Gym就能搭建出来很适合作为AI创造公开赛的参赛作品也很适合作为强化学习入门项目深入研究。希望这篇文章能给你一个比较完整的“AI交易游戏”实现思路。接下来你可以打开编辑器先跑通数据生成再进环境定义一步步让AI从一个只会随机买卖的新手成长为能控制回撤的猎手。