资讯动态

Python全自动量化交易系统:数据、回测与实盘执行链路

发布时间:2026/10/30 2:50:00 来源:尧图企业网站定制
简介面向希望用 Python 搭建全自动交易系统、实施量化策略的投资者与开发者这份资料围绕金融数据处理、技术指标计算、回测验证与实盘执行展开适合具备 Python 基础、想系统串联算法交易流程的学习者。压缩包共 35 个文件以 34 个 py 脚本和 1 个 md 说明为主整体约 38KB脚本覆盖数据获取、组合再平衡、夏普与索提诺比率、回撤与波动率、RSI、MACD、ADX、布林带、Piotroski F 分数、神奇公式及可视化等模块。已有 3962 人学习下载。读者可从中获得可运行的策略研究代码片段、常见指标与绩效评价实现、回测与风险控制思路以及连接行情接口和整理交易日志的参考做法便于按模块拆解学习并组合成自己的量化交易原型。1. 全自动交易系统真正的难点链路难接策略相对容易很多人接触 Python 算法交易都是从一条均线交叉开始的行情从 CSV 读进来信号打出来回测曲线漂亮得不像话。等到要把这套东西做成全自动交易系统才发现真正耗时间的地方在别处行情时间戳对不齐、复权因子漏了、信号按收盘价成交、重连之后重复下单定量交易策略本身反而没那么难。这份 Python 量化项目把「数据 → 因子 → 策略 → 回测 → 实盘执行」整条链路都摆在一套目录里适合两类人写过几行 pandas、想把研究代码变成能每天自动跑的交易系统的开发者以及有策略想法、但还没搭过完整工程骨架的人。下面的顺序和我自己拆包时一致先看它长什么样再看关键代码最后说怎么改。2. 工程骨架目录结构、依赖清单与配置怎么摆拿到一套交易系统源码多数人第一反应是打开策略文件看用了什么指标。我更习惯先看目录和配置因为策略赚不赚钱有点玄学工程结构规不规整却是一眼能看出来的。一个能长期跑的全自动交易系统和一份跑完就扔的研究脚本差别基本都在骨架里能不能重启、能不能换券商、能不能在不改代码的前提下调参数。先说结论这套项目的分层思路是「研究区」和「生产区」物理隔开行情、策略、风控、撮合四块各自独立成包互相之间只通过数据结构通信。这个约束听起来很教条但它决定了你后面加策略是改三个文件还是改三十个文件。2.1 一套目录骨架把研究代码和生产代码隔开拆开压缩包主目录大致是这样一棵树不同版本可能有增删主干跑不掉quant-trader/ ├── config/ │ ├── settings.yaml # 全局配置账户、费率、风控阈值 │ └── strategies/ # 每个策略一份参数文件 ├── data/ │ ├── raw/ # 接口原始返回按日期落盘出了问题能回溯 │ ├── processed/ # 清洗 复权后的结果回测直接读 │ └── market.db # 本地行情库SQLite ├── core/ │ ├── context.py # 策略上下文当前 bar、账户、持仓 │ ├── portfolio.py # 资金、持仓、成本、盈亏结算 │ └── events.py # 信号 / 订单 / 成交的数据结构 ├── datafeed/ │ ├── loader.py # 行情采集屏蔽不同数据源的差异 │ ├── cleaner.py # 清洗、复权、停牌标记 │ └── factors.py # 因子计算纯函数 ├── strategy/ │ ├── base.py # 策略基类规定接口 │ ├── dual_ma.py # 双均线示例 │ └── breakout.py # 通道突破示例 ├── risk/ │ ├── rules.py # 交易前风控检查 │ └── sizer.py # 目标权重 → 股数换算 ├── broker/ │ ├── gateway.py # 撮合网关抽象基类 │ ├── sim_gateway.py # 模拟撮合回测和模拟盘共用 │ └── live_gateway.py # 实盘适配层换券商只改这一个文件 ├── backtest/ │ ├── runner.py # 回测主循环 │ ├── metrics.py # 绩效统计 │ └── walkforward.py # 滚动样本外验证 ├── live/ │ ├── scheduler.py # 定时任务盘前、盘中、盘后 │ └── recover.py # 断线重连、状态恢复 ├── logs/ # 结构化日志按日切分 ├── main.py # 唯一入口靠 --mode 切换回测/模拟/实盘 └── requirements.txt几个目录的职责值得单独说。core/是纯内存对象不 import 任何数据源和券商 SDK这样你在本地拿一份 CSV 就能把账户逻辑跑通broker/用抽象基类把「下单」这件事收口sim_gateway和live_gateway实现同一套方法签名回测和实盘才能共用策略代码strategy/里严禁出现portfolio.cash ...这种语句策略只能吐信号钱怎么动是core/portfolio.py的事。这条规矩在项目初期会觉得多余等你要同时维护五个策略的时候就明白它是后悔药。main.py只做一件事读配置、装配对象、把控制权交给对应的 runner。装配逻辑集中在一处排查问题时不用满仓库找入口。2.2 依赖与环境能少一个是一个这套项目的依赖相当克制核心就三样数据处理用 pandas / numpy配置用 PyYAML日志用标准库 logging 或一个轻量封装。定时调度部分用 APScheduler 这类进程内调度器就够了不需要引入消息队列和分布式框架——个人或小团队的账户规模一台机器一个进程完全撑得住。血泪经验是不要一上来就选那种「一站式量化框架」。它们把回测、撮合、指标全封好了看着省事可一旦回测和实盘结果对不上你就得钻进别人的黑匣子里去对撮合假设。这套项目自己写回测循环虽然只有几百行但每一行的假设你都看得见、改得动。环境隔离和启动命令很朴素# 建虚拟环境避免和系统里的 pandas 版本打架 python -m venv .venv source .venv/bin/activate # Windows 用 .venv\Scripts\activate pip install -r requirements.txt # 三个模式共用同一个入口先跑回测再跑模拟盘最后才是实盘 python main.py --config config/settings.yaml --mode backtest python main.py --config config/settings.yaml --mode paper--mode这个参数是整套设计里最值钱的一行。它保证backtest、paper、live三种运行方式走的是同一套策略代码、同一套风控代码唯一被替换的是broker/下的网关实现。很多人回测和实盘用两份代码结果两边行为不一致对账对到怀疑人生。requirements.txt建议锁死小版本号比如pandas2.1.*这种写法行情数据只有一份跑出结果之后再升级依赖等于把历史回测结论全部作废一次。2.3 配置外置账户信息和策略参数别写死在代码里策略参数外置是最基础的一条。除此之外账户号、接口地址这类东西绝对不能进版本库常见做法是用环境变量占位符。配置长这样# config/settings.yaml mode: backtest # backtest | paper | live universe: [600000, 000001] # 标的池实盘前先确认代码格式和交易所后缀 bar: 1d # 1d 日线1m 分钟线 data: raw_dir: data/raw db_path: data/market.db cost: commission_bps: 2.5 # 双边佣金按万分之 2.5 估 stamp_tax_bps: 5 # 卖出印花税只在卖出方向收 slippage_bps: 3 # 滑点假设回测里必须显式打开 risk: max_position_pct: 0.2 # 单票最大仓位占比 max_drawdown_stop: 0.15 # 账户回撤熔断线 max_orders_per_min: 6 # 下单频率上限防止逻辑写错刷单 broker: gateway: sim # sim | live account_id: ${ACCOUNT_ID} # 从环境变量注入不落盘 endpoint: ${BROKER_ENDPOINT}读取配置的代码顺带做两件事展开占位符、校验必填项。# core/config.py import os import re from pathlib import Path import yaml _ENV_PATTERN re.compile(r\$\{(\w)\}) def _expand(value): 把 ${ENV} 占位符替换成环境变量找不到就留空串由后面的校验兜住 if isinstance(value, str): return _ENV_PATTERN.sub(lambda m: os.environ.get(m.group(1), ), value) if isinstance(value, dict): return {k: _expand(v) for k, v in value.items()} if isinstance(value, list): return [_expand(v) for v in value] return value def load_config(path: str config/settings.yaml) - dict: raw yaml.safe_load(Path(path).read_text(encodingutf-8)) cfg _expand(raw) # 必填项校验宁可启动就崩也别带着空账户号跑到实盘 missing [k for k in (mode, universe, data, cost, risk) if not cfg.get(k)] if missing: raise ValueError(f配置缺少必填项: {missing}) if cfg[mode] live and not cfg[broker].get(account_id): raise ValueError(live 模式下 ACCOUNT_ID 未注入) return cfg_expand递归处理 dict 和 list因为实盘配置里经常有嵌套结构比如多个账户。真正关键的是最后那个判断mode为live而账户号是空串时直接抛异常。我见过太多次「本地调试忘了改配置一启动连到了模拟环境还以为是实盘没成交」的乌龙加三行校验比事后查日志便宜得多。3. 数据与因子从原始行情到一张可回测的特征表数据层决定了这套系统八成以上的可信度。策略写得再花喂进去的行情少了复权、混了停牌日、时间戳差了八小时回测出来的年化就是个安慰剂。这一层的目标很单一不管上层用的是日线还是分钟线、来自哪个数据源最终都输出一张列名固定、带复权价、带停牌标记的宽表。接口返回格式各家不一样datafeed/loader.py的职责就是把这层差异吃掉——列名统一成小写时间统一成带时区的时间戳然后落一份原始文件到data/raw/。留原始文件不是洁癖是为了在发现清洗逻辑写错时能重放不用重新拉一遍历史。3.1 清洗与复权先保证每一根 K 线是真的清洗函数不长但每条规则背后都有具体的翻车场景。# datafeed/cleaner.py import pandas as pd REQUIRED [dt, symbol, open, high, low, close, volume] def clean_bars(df: pd.DataFrame) - pd.DataFrame: 把接口原始返回整理成规整的 K 线表 df df.copy() df.columns [c.strip().lower() for c in df.columns] # 1) 接口偶尔返回半截数据缺关键列的整行丢掉不要试图补 df df.dropna(subset[c for c in REQUIRED if c in df.columns]) df[dt] pd.to_datetime(df[dt]) # 2) 排序 去重同一根 K 线被重复拉取是常态保留最后一次 df df.sort_values([symbol, dt]) df df.drop_duplicates([symbol, dt], keeplast) # 3) 价格逻辑校验最高价小于最低价的整根丢弃 df df[df[high] df[low]] # 4) 停牌日成交量为 0数据保留但打标记回测里不许在这里成交 df[suspended] df[volume] 0 return df.reset_index(dropTrue) def adjust_price(df: pd.DataFrame, factors: pd.DataFrame) - pd.DataFrame: 前复权把历史价格拉到最新价口径算指标必须用复权价 df df.merge(factors[[symbol, dt, adj_factor]], on[symbol, dt], howleft) # 复权因子按标的前向填充缺失时退回 1.0避免整段数据变成 NaN df[adj_factor] df.groupby(symbol)[adj_factor].ffill().fillna(1.0) for col in (open, high, low, close): df[f{col}_adj] df[col] * df[adj_factor] return df几点说明。suspended这一列不是装饰回测撮合时必须查它否则除权除息日或者长期停牌复牌那天会出现「按停牌前的价格买到了停牌期的票」这种不可能成交的记录。复权用前复权而不是后复权是因为下单要用最新价格信号计算和下单价格必须在同一口径下中间只差一个adj_factor的换算——这个换算如果放在策略里做迟早会写错一次。adj_factor缺失时退回 1.0 而不是直接 drop是为了保住行情连续性但这件事要在日志里记一笔因为因子缺失往往意味着数据源那边有问题属于需要人工确认的异常。3.2 因子计算写成纯函数且统一 shift(1)因子层最容易被忽略的问题是对未来函数处理不当。所谓未来函数就是你在计算今天这根 K 线的信号时不小心用到了今天收盘之后才知道的信息。回测里它会送你一条漂亮到不真实的曲线实盘里它会让你怀疑人生。这套项目里所有因子统一用shift(1)也就是今天产生的信号只能用昨天及之前的数据# datafeed/factors.py import pandas as pd def add_features(df: pd.DataFrame, fast: int 5, slow: int 20, atr_n: int 14) - pd.DataFrame: 给 K 线表补上均线、ATR、收益 z 分数全部按标的独立计算 df df.sort_values([symbol, dt]).copy() g df.groupby(symbol, group_keysFalse) # 均线先在组内 rolling再整体 shift(1)保证不含当日收盘价 df[ma_fast] g[close_adj].transform(lambda s: s.rolling(fast).mean()).shift(1) df[ma_slow] g[close_adj].transform(lambda s: s.rolling(slow).mean()).shift(1) # ATR真实波幅取三者最大用来做止损距离 prev_close g[close_adj].shift(1) tr pd.concat([ df[high_adj] - df[low_adj], (df[high_adj] - prev_close).abs(), (df[low_adj] - prev_close).abs(), ], axis1).max(axis1) df[atr] tr.groupby(df[symbol]).transform( lambda s: s.rolling(atr_n).mean()).shift(1) # 20 日收益的滚动 z 分数用来过滤短期过热 ret g[close_adj].pct_change() df[ret_z] ret.groupby(df[symbol]).transform( lambda s: (s - s.rolling(20).mean()) / (s.rolling(20).std() 1e-9) ).shift(1) # 前 slow 根没有完整窗口直接丢掉别用 NaN 参与判断 return df.dropna(subset[ma_slow, atr])三个参数各有作用fast/slow是均线窗口直接决定信号频率日线下 5/20 意味着大约一个月换手一次atr_n是波动率窗口用来把止损距离设成「几倍 ATR」这比固定百分比止损适应性强得多波动大的标的自动给宽一点的止损。这里有一个容易翻车的细节transform之后再整体shift(1)和先shift(1)再rolling的结果不一样。前者是「昨天的均线值」后者是「包含昨天在内的均线值」。项目里取的是前者因为策略在开盘前跑能拿到的只有到昨天收盘为止的完整信息。如果你的策略是收盘前几分钟决策口径要相应调整但一定要在文档里写清楚否则半年后自己都说不清这条曲线是怎么来的。3.3 落库与增量更新让回测从 20 分钟变成 20 秒因子算完不落库每次回测都重算一遍数据量一上来就很难受。项目用 SQLite 存日线表结构和增量写入逻辑都不复杂-- data/schema.sql CREATE TABLE IF NOT EXISTS bars_daily ( symbol TEXT NOT NULL, dt TEXT NOT NULL, -- ISO 字符串时区口径全库统一禁止混用 open REAL, high REAL, low REAL, close REAL, volume REAL, adj_factor REAL DEFAULT 1.0, suspended INTEGER DEFAULT 0, -- 0/1SQLite 没有布尔类型 PRIMARY KEY (symbol, dt) ); CREATE INDEX IF NOT EXISTS idx_bars_dt ON bars_daily (dt);# datafeed/store.py import sqlite3 import pandas as pd def upsert_bars(db_path: str, df: pd.DataFrame) - int: 按 (symbol, dt) 主键覆盖写入重复拉取不会产生脏数据 cols [symbol, dt, open, high, low, close, volume, adj_factor, suspended] rows df[cols].assign(dtdf[dt].dt.strftime(%Y-%m-%d %H:%M:%S)).values.tolist() with sqlite3.connect(db_path) as conn: conn.executemany( INSERT OR REPLACE INTO bars_daily (symbol, dt, open, high, low, close, volume, adj_factor, suspended) VALUES (?,?,?,?,?,?,?,?,?), rows) return len(rows) def load_bars(db_path: str, symbols: list[str], start: str, end: str) - pd.DataFrame: 回测统一从这里取数禁止各模块自己连库拼 SQL sql (SELECT * FROM bars_daily WHERE symbol IN ({}) AND dt BETWEEN ? AND ? ORDER BY symbol, dt) placeholders ,.join(? * len(symbols)) with sqlite3.connect(db_path) as conn: return pd.read_sql(sql.format(placeholders), conn, params[*symbols, start, end])INSERT OR REPLACE配合(symbol, dt)主键天然就是幂等的——今天补拉了三天数据跑两遍结果一样。增量更新时只需要记住每个标的的最后一条dt从那天往后拉不用全量重跑。标的数量在几百以内SQLite 完全够用真到几千只股票加分钟线再换列式存储也不迟而且因为取数统一走load_bars换存储只改这一个函数。提示这里所有的时间字段都要求同一个时区口径。项目约定库里统一存交易所本地时间展示层再转混用是最常见的排查噩梦。4. 策略与回测引擎一个信号从产生到资金结算的完整链路到这一层代码开始有「系统」的样子了。一根 K 线进来先由策略产生信号信号经风控换算成订单订单交给网关撮合成交回写到账户账户再算出新的净值。这五个环节在代码里必须是五段独立逻辑而不是揉在一个for循环里——揉在一起的回测脚本你永远不知道最后那个数字是策略赚的还是撮合 bug 送的。4.1 策略基类只允许吐信号不许碰账户基类定义得很薄就两个方法但把约束写死了# strategy/base.py from dataclasses import dataclass dataclass class Signal: symbol: str action: str # buy | sell weight: float # 目标权重不是股数 reason: str # 信号原因码用于事后归因 class BaseStrategy: name base params: dict {} def prepare(self, df): 一次性算好因子回测和实盘共用同一份逻辑 raise NotImplementedError def on_bar(self, ctx) - list[Signal]: 每根 K 线调用一次只允许返回信号禁止修改账户 raise NotImplementedErrorweight是目标权重而不是股数这一条很关键。策略不该知道账户里有多少钱、一手是多少股这些是风控和撮合的事。权重接口带来的直接好处是同一个策略账户规模从十万变到一千万代码一行不用改。下面这个双均线策略是项目自带的示例二十几行把该有的东西都摆出来了# strategy/dual_ma.py from strategy.base import BaseStrategy, Signal from datafeed.factors import add_features class DualMA(BaseStrategy): name dual_ma params {fast: 5, slow: 20, atr_stop: 2.0} def prepare(self, df): return add_features(df, self.params[fast], self.params[slow]) def on_bar(self, ctx): row ctx.bar pos ctx.position(row.symbol) out [] # 金叉且空仓 → 建仓到目标权重 if row.ma_fast row.ma_slow and pos.qty 0: out.append(Signal(row.symbol, buy, 0.2, reasonma_cross_up)) # 死叉且持仓 → 清仓 elif row.ma_fast row.ma_slow and pos.qty 0: out.append(Signal(row.symbol, sell, 0.0, reasonma_cross_down)) # 价格跌破 成本 - N 倍 ATR → 止损独立于均线信号 elif pos.qty 0 and row.close_adj pos.cost - self.params[atr_stop] * row.atr: out.append(Signal(row.symbol, sell, 0.0, reasonatr_stop)) return outreason字段平时看着多余等你要回答「这个月亏的钱是均线信号亏的还是止损砍的」时它就是唯一能给你答案的东西。所有成交记录里都带上原因码盘后按原因聚合一下策略哪一部分在拖后腿一目了然。atr_stop设成 2.0意思是止损距离等于两倍 ATR。这个值调大止损更宽松、交易次数更少调小回撤控制更紧但容易被正常波动洗出去。没有普适最优值通常的做法是固定在 1.53 之间然后在滚动样本外验证里看它对结果的影响是否剧烈——如果参数动 0.2 结果就天翻地覆那这个策略本身就不稳。4.2 回测循环把撮合假设明明白白写在代码里回测主循环是整套项目里我最愿意逐行读的部分因为它把所有假设都暴露出来了# backtest/runner.py def run_backtest(strategy, bars, cfg, start_cash: float 1_000_000.0): 按交易日推进先撮合昨天收盘后产生的信号再让策略产生新信号 portfolio Portfolio(cashstart_cash, costcfg[cost]) ctx Context(portfolio) slip cfg[cost][slippage_bps] / 1e4 for dt, day_bars in bars.groupby(dt): # 1) 用今天开盘价成交昨天盘后的信号这是最贴近实际的假设 for sig in ctx.pending_signals: bar day_bars[day_bars[symbol] sig.symbol] if bar.empty or bar.iloc[0][suspended]: continue # 停牌没法成交直接作废这一笔 px bar.iloc[0][open_adj] * (1 slip if sig.action buy else 1 - slip) portfolio.execute(sig, px, dt) ctx.pending_signals [] # 2) 逐根更新上下文让策略基于今天的数据产生明天的信号 for _, row in day_bars.iterrows(): ctx.bar row ctx.pending_signals.extend(strategy.on_bar(ctx)) # 3) 收盘结算净值写进曲线 portfolio.mark_to_market(day_bars) return portfolio.history这段代码里有三个必须讲清楚的假设。第一信号和成交错开一天昨天的信号用今天的开盘价成交这是为了避免「用当天收盘价成交」这个经典错误——收盘价要收盘才知道实盘根本拿不到。第二滑点按方向加买入往上加、卖出往下减很多人只在买入方向算滑点等于凭空多赚一笔。第三停牌直接跳过而不是顺延顺延会让回测里多出一批实盘不可能有的成交。Portfolio.execute里还有几条规则藏在细节中卖出时要额外扣印花税要按照 T1 判断当日买入的股数能不能卖要按 100 股一手向下取整。这几条任何一条漏了回测结果都会偏乐观而偏乐观的回测是最危险的——它不会让你亏钱它会让你以为自己能赚钱。提示回测数字只反映历史区间和当前成本假设别把它当成未来收益的预期。回测跑通之后建议把slippage_bps从 3 调到 10 再跑一遍。如果策略收益在这两档之间崩塌说明它赚的是摩擦成本的钱实盘大概率复制不了。4.3 绩效指标别只盯年化收益绩效统计的代码短但看哪些指标决定了你会不会做出错误判断# backtest/metrics.py import numpy as np import pandas as pd def summarize(equity: pd.Series, trades: pd.DataFrame, periods: int 252) - dict: ret equity.pct_change().dropna() ann (1 ret).prod() ** (periods / len(ret)) - 1 # 年化收益 dd (equity / equity.cummax() - 1).min() # 最大回撤负数 sharpe ret.mean() / (ret.std() 1e-12) * np.sqrt(periods) win (trades[pnl] 0).mean() if len(trades) else 0.0 return { annual_return: round(ann, 4), max_drawdown: round(dd, 4), sharpe: round(sharpe, 2), calmar: round(ann / abs(dd), 2) if dd else None, # 收益/回撤 win_rate: round(win, 4), turnover: round(trades[notional].sum() / equity.mean(), 2), trade_count: int(len(trades)), }各指标该看什么列一下更直观指标计算口径什么时候要警惕annual_return复利折算到 252 个交易日单看它没有意义max_drawdown净值相对历史高点的最大跌幅超过风控阈值就该否掉策略sharpe日收益均值 / 标准差年化大于 2 且换手极低先怀疑未来函数calmar年化收益 / 最大回撤比夏普更贴近真实体感win_rate盈利笔数 / 总笔数40% 胜率配 2:1 盈亏比是正常的turnover成交金额 / 平均净值超过 20 倍/年手续费会吃掉大部分收益最容易被忽略的是turnover。日频策略一年换手三四十倍按万分之三滑点加万分之七综合成本算光摩擦就要吃掉三四个点的年化。策略在回测里赚 8%成本敏感度测一遍可能就剩 4% 了这个落差必须提前知道。5. 实盘执行与排查把信号搬到真实撮合上最容易翻车的五个点回测跑通只是及格线。真正让一套全自动交易系统能不能上线的是执行层——订单状态怎么跟踪、断线了怎么恢复、风控在哪一层拦。这部分代码量不大但每一行都关系到资金安全所以我把它和排查经验放在同一章讲。5.1 网关抽象与订单状态机网关的定义尽量贴近真实券商接口的形态连接、查持仓、查未结订单、下单、撤单、成交回调。# broker/gateway.py from enum import Enum class OrderStatus(Enum): PENDING pending # 本地已生成尚未发出 SUBMITTED submitted # 已发到柜台等回执 PARTIAL partial # 部分成交 FILLED filled CANCELED canceled REJECTED rejected # 被柜台或风控拒绝 class Gateway: 所有券商对接都实现这套方法签名回测/模拟/实盘才能共用上层代码 def connect(self): ... def query_positions(self) - dict: ... def query_open_orders(self) - list: ... # 重连后第一件事就是调它 def submit(self, order) - str: ... # 返回 client_order_id def cancel(self, client_order_id: str): ... def on_fill(self, callback): ...client_order_id是客户端自己生成的唯一键格式常见做法是{日期}-{标的}-{当日序号}。它的意义只有一条幂等。网络超时的时候你不知道订单到底发出去没有重发一次可能变成双倍仓位不重发可能漏单。有了这个 ID重发前先查未结订单里有没有同一个 ID问题就收敛了。状态机里要特别注意PARTIAL。订单部分成交是常态尤其是流动性一般的标的如果你只在FILLED时才更新持仓账户和柜台就会对不上。正确做法是成交回调里就按实际成交数量累加PARTIAL之后继续等后续回调直到终态。5.2 交易前风控闸门所有订单必须过一遍风控函数不做交易决策只做否决。它的位置在所有订单发给网关之前唯一出口# risk/rules.py def pre_trade_check(order, account, cfg) - list[str]: p a hrefhttps://download.csdn.net/download/weixin_42097533/15279118 stylecolor:#ec7500;font-size:14px; 本文还有配套的精品资源点击获取 /a img altmenu-r.4af5f7ec.gif srchttps://csdnimg.cn/release/wenkucmsfe/public/img/menu-r.4af5f7ec.gif stylewidth:16px;margin-left:4px;vertical-align:text-bottom;cursor:text; /p

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价 →
↑