简介：这是一份面向Python GUI开发初学者与进阶者的PyQt实战课程资源，核心围绕用PyQt搭建视频监控系统展开，将界面设计、多媒体播放、摄像头取流与后台处理串成可运行的完整项目案例。内容覆盖窗口与布局管理、常用控件与布局器、QMediaPlayer与QVideoWidget视频播放、OpenCV摄像头接入、多线程分离、信号与槽通信机制、事件处理、Qt Designer可视化设计，以及视频录制回放和网络传输等关键环节，帮助学习者从基础控件使用逐步过渡到综合桌面应用开发。压缩包共28个文件，以Python源码、PNG界面素材、XML配置和PYC缓存为主，整体约84KB，并保留完整项目代码、图标资源与IDE配置等相关目录结构，便于按模块阅读、运行和调试。已有358人学习，可作为课程配套案例，用于理解视频监控系统的整体组织方式、模块划分与实现思路。

1. 用 pyq 实战教程把策略从脑补搬到回测曲线上

前阵子有个做趋势策略的朋友发来一张回测图，年化六十多，最大回撤四个点。我问他成交价怎么取的，他想了三秒说：收盘价。这句话基本就给那条曲线判了刑。用 pyq 实战教程这套课程资源走一遍，最值钱的不是学会几个指标怎么调，而是能指出哪一行假设正在偷偷替你赚钱。

pyq 是一套 Python 写的事件驱动量化回测框架，按 bar 推进，自带撮合、费用与绩效统计。这套教程把它拆成四段：数据接入、策略编写、参数与成本建模、多标的事件驱动，每段都配可运行示例。

适合已经会写 pandas、但没搭过完整回测链路的开发者，也适合策略能跑出来、却说不清收益来源的人。不适合想直接拿买卖信号的人，这里没有那种东西。

2. pyq 的骨架：行情、事件、撮合三段链路怎么串

2.1 事件驱动与向量化的取舍：什么时候必须换引擎

大多数人的第一次回测是 pandas 里两行 rolling mean，shift 一下算收益，cumprod 出曲线，五分钟能出结果。这种向量化写法的问题在于，它把“时间”压成了一个数组下标，今天和昨天的先后关系只剩下索引差 1。一旦策略里出现“跌破成本价就止损”“同一天两只标的同时给信号但账上只有一份钱”，向量化就开始靠 shift 硬凑，凑出来的结果自己都不敢信。

pyq 走的是事件驱动。引擎维护一条按时间排序的事件队列，bar 到达触发on_bar，策略发出订单，撮合模块按当前 bar 的成交假设判断能否成交、成交多少、扣多少费用，回报再回写持仓与资金。链路长了，换来的是每个环节都可查——这条订单为什么没成交、这笔钱从哪一刻开始被占用，日志里都留痕。

代价要讲清楚：事件驱动比向量化慢一到两个数量级，代码量翻倍，调试从“看数组”变成“看时序”。我的判断标准很简单——单标的、日频、只做多、满仓进出，向量化够用；只要出现多标的共享资金、盘中止损、分批建仓、部分成交中的任意一条，就该换事件驱动。这套教程的示例基本都落在后者，这也是它比一般指标教学更值得花时间的地方。

还有个隐性好处：事件驱动的策略代码和实盘接口长得很像。回测里on_bar收到的是一个 bar 对象，换成实时行情推送，结构不用重写。我见过太多团队向量化回测跑得飞起，接实盘时整个策略层推倒重来，那才是最贵的一课。

2.2 数据先标准化：字段、复权、停牌三件事

拿到行情第一件事不是写策略，是把数据整理成引擎认的 schema。我见过太多“回测跑不通”，最后定位到 CSV 里有一列叫Close有一列叫close，或者某天缺了一行，导致整条均线窗口前移一天，收益好看得莫名其妙。

import pandas as pd def load_bars(path, symbol): df = pd.read_csv(path) df.columns = [c.strip().lower() for c in df.columns] # 时间戳统一到 naive datetime，时区在入库前就剥掉，别留到引擎里 df["dt"] = pd.to_datetime(df["dt"], format="%Y-%m-%d %H:%M:%S") df = df.sort_values("dt").drop_duplicates("dt") # 复权必须显式做：引擎不做隐式复权，避免同一份数据被复权两次 if "adj_factor" in df.columns: for col in ("open", "high", "low", "close"): df[col] = df[col] * df["adj_factor"] df = df.set_index("dt") # 停牌日不能直接删行，删了均线窗口会跨过停牌期，等于偷看未来 full_idx = pd.date_range(df.index.min(), df.index.max(), freq="1D") df = df.reindex(full_idx) df["volume"] = df["volume"].fillna(0) # 停牌日成交量为 0 df[["open", "high", "low", "close"]] = df[["open", "high", "low", "close"]].ffill() df["symbol"] = symbol return df

这段代码的重点不在语法，在三个决定：adj_factor自己乘、停牌日补行、成交量填 0。第一条保证复权只发生一次；第二条保证滚动窗口按自然交易日推进，而不是按“有行情的日子”推进；第三条给后面的撮合留了判断依据——成交量为 0 的 bar，任何订单都不该成交。

参数上没什么可调的，但有几个字段必须确认。dt的格式串要和源文件严格一致，写成%Y/%m/%d而文件里是横杠，to_datetime会静默解析失败或者整体报错；freq="1D"只在日频成立，分钟频要换"1min"，且必须先确认原始数据里有没有集合竞价那几根；ffill()只能用在价格列，绝不能用在前复权因子上，因子断了就是断了，补出来的价格会失真。

如果源数据本身是后复权价，就不要再乘因子，否则等于把复权做了两遍，早期价格会被压到接近零，回测里会出现“涨了几十倍”的幻觉。判断方法很土但有效：随便挑一只除权过的票，看它除权日前后价格有没有跳空，跳了就是未复权。

2.3 引擎启动：四个参数定死回测的边界

数据齐了，下一步是把引擎拉起来。示例里的类名和参数名按我习惯的写法给，资源包里的命名可能差一两个字母，对着改字段不费事。

from pyq import Engine, Context from strategies import DoubleMA ctx = Context( cash=1_000_000, # 初始资金，决定手数取整后的实际仓位偏差 commission_rate=0.00025, # 双边佣金，卖出时另有印花税需单独加 min_commission=5.0, # 单笔最低佣金，小资金回测里这条最容易被忽略 slippage_ticks=1, # 滑点按最小变动价位的倍数计 price_limit=0.10, # 涨跌停幅度，用于拒单判断 ) eng = Engine(context=ctx, freq="1d") eng.add_data(load_bars("data/000001.csv", "000001")) eng.add_strategy(DoubleMA(fast=20, slow=60)) eng.run(start="2020-01-01", end="2024-12-31")

cash不是随便填的。十万本金配 5 元最低佣金，等于每次交易固定抽走 0.005%，一天来回两次就是 1%，高频一点的策略光佣金就能把收益吃干净。我一般先用真实资金量级跑一遍，再放大到一百万看策略本身的超额收益，两个结果差得离谱，说明策略对资金规模敏感，实盘上量要谨慎。

slippage_ticks是这套参数里最容易被调小的一个。设成 0，回测立刻好看一截，但那是在假设你能按收盘价精确成交。日频策略我一般给 1 到 2 个 tick，分钟频给 2 到 3 个，商品期货按最小变动价位折算后会更狠。

freq要和数据粒度对齐，日频数据喂给分钟引擎，引擎会按天推进但策略里的on_bar被调用次数完全不对，日志里看成交量就能发现异常。start/end建议比你想回测的区间各多留一段，前面的用于指标预热，后面的用于持仓清算，不然首尾两段的绩效指标都是残的。

3. 第一个可回测策略：把信号、下单、风控拆成三段代码

3.1 为什么必须拆成三段

新手写策略最常见的写法是把判断、下单、止损全塞在on_bar里，一个函数两百行，改个参数要通读一遍。这种代码在单标的、单策略时还能忍，一旦要在同一份资金上跑三个策略，就没法定位问题——到底是信号错了，还是仓位算错了，还是止损提前触发了。

我一般按三段切：信号段只输出“想做多、想做空、想平”的意图，不碰钱；下单段把意图翻译成目标仓位，再换算成手数；风控段在订单发出前做最后一道闸，管单标的上限、日内亏损、单笔止损。三段之间用明确的函数边界隔开，出了问题看日志就知道卡在哪一段。

这么拆还有个附带好处：同一套风控可以挂在多个信号上，同一套信号可以换不同的下单逻辑做对比。教程里的示例基本是这个结构，照着改比自己重新设计省事。

3.2 信号段：只用收盘之后能看到的信息

class DoubleMA(Strategy): params = dict(fast=20, slow=60, max_pos=0.30, stop_loss=0.08) def on_bar(self, bar, ctx): # 只取到当前 bar 为止的历史，最后一个元素是刚收盘的这根 hist = ctx.history(bar.symbol, self.params["slow"] + 2) if len(hist) < self.params["slow"] + 1: return # 预热期不足，直接跳过 close = hist["close"] fast_ma = close.tail(self.params["fast"]).mean() slow_ma = close.tail(self.params["slow"]).mean() prev_fast = close.iloc[-self.params["fast"] - 1:-1].mean() prev_slow = close.iloc[-self.params["slow"] - 1:-1].mean() # 用"昨天没叉、今天叉了"来判定，避免连续多日重复发信号 golden = prev_fast <= prev_slow and fast_ma > slow_ma death = prev_fast >= prev_slow and fast_ma < slow_ma if golden: self.set_target(bar.symbol, self.params["max_pos"]) elif death: self.set_target(bar.symbol, 0.0)

ctx.history的第二个参数是“要多少根”。这里取slow + 2，一根给当前 bar，一根给 prev 计算用的窗口，少取一根就会在交叉那天拿到不完整的均值，信号直接错位一天。这种错位在图上完全看不出来，但收益能差十几个点。

golden/death用前后两期比较，是为了避免“快线一直在慢线上方”时每天都发一次做多信号。很多人的回测换手率高得离谱，就是少了这个判断。另外用<=和<混着写，是为了处理两线相等那一天，这种边界不处理，交叉日会漏或者重复。

max_pos=0.30是目标仓位占净值的比例，不是手数。用比例表达的好处是资金曲线涨了仓位自动跟上，不用手动改手数；坏处是回撤时仓位缩小，反弹时参与度不足。这是设计取舍，不是 bug。

3.3 下单段：目标仓位比“买几手”好维护

def set_target(self, symbol, target_ratio): price = self.ctx.last_price(symbol) equity = self.ctx.equity # 总资产 = 现金 + 持仓市值 target_value = equity * target_ratio target_shares = int(target_value / price / 100) * 100 # A 股按手取整 current = self.ctx.position(symbol).shares delta = target_shares - current if delta == 0: return side = "buy" if delta > 0 else "sell" qty = abs(delta) # 买入前用可用资金复核，别让引擎替你兜底 if side == "buy": need = qty * price * (1 + self.ctx.commission_rate) if need > self.ctx.available_cash: qty = int(self.ctx.available_cash / price / 100) * 100 if qty <= 0: return self.ctx.order(symbol, qty, side)

取整这一步是必须的。A 股按 100 股一手，算出来 1234 股实际只能下 1200 股，直接把 34 股丢给引擎，要么被拒单，要么被向下取整后剩个零头，日志里会看到一堆“下单失败”。期货更麻烦，不同品种合约乘数不同，得先查合约信息再除。

self.ctx.available_cash和equity不是一个东西。前者是扣掉冻结资金后能立刻用的钱，后者包含持仓市值和浮动盈亏。用equity算目标仓位、用available_cash复核能不能买，这两个数混用，多标的回测里必然出现资金超发——账上只有十万，三个策略各下了五万。

order之后不要假设一定成交。撮合在下一个环节，滑点、涨跌停、成交量不足都可能让订单部分成交或全废。策略里对成交的依赖要通过持仓查询获得，而不是下单成功就当成了。

3.4 风控段：三条硬约束写进代码而不是脑子

def risk_check(self, symbol, delta_qty, price, ctx): pos = ctx.position(symbol) equity = ctx.equity # 1) 单标的名义市值上限，防止目标仓位算错导致满仓单票 new_value = (pos.shares + delta_qty) * price if new_value > equity * 0.35: return False # 2) 单笔止损：浮亏超过阈值先平掉，不等信号反转 if pos.shares > 0 and price < pos.cost * (1 - self.params["stop_loss"]): ctx.order(symbol, -pos.shares, "sell") return False # 3) 日内熔断：当日净值回撤超 3% 当天不再开新仓 if ctx.drawdown_today() > 0.03 and delta_qty > 0: return False return True

第一条是保命的。目标仓位算错、除零、复权跳变都可能让max_pos失效，硬上限能兜住最坏情况。第二条把止损从“信号”里拿出来单独做，因为趋势策略的信号天然滞后，等死叉再走，回撤早就超了。第三条是日内熔断，日频回测里作用有限，但一旦换成分钟频或者多标的组合，它能挡住连环爆仓。

三条约束的阈值不要一起调。我一般固定两条、只调一条，调完看收益变化是来自收益提升还是回撤放大。三条一起动，最后根本说不清是哪条起的作用。

4. 参数与绩效：把回测结果从“好看”固定到“可信”

4.1 三个成本参数：手续费、滑点、成交价假设

成本建模这块，多数教程一笔带过，实际它决定了回测和实盘的差距量级。佣金、印花税、过户费这些是明面上的，按费率加就行；滑点是隐性的，取决于你的下单方式和标的流动性；成交价假设最致命，因为它直接决定了信号到成交的那一跳。

日频策略我一般这么设：信号在收盘后产生，成交按次日开盘价加一个 tick 的滑点。理由很简单，你收盘时看到的数据，最早也只能在次日开盘前下单。教程示例里如果默认按当日收盘价成交，记得改掉，这一处改动通常会让年化掉十几个点，但那才是真实的。

分钟频还要考虑冲击成本，尤其是小市值标的。一个简单的近似：成交价 = 开盘价 × (1 + k × 下单量 / 该 bar 成交量)，k 取 0.1 到 0.3。这个公式不精确，但比不设强得多。

4.2 参数网格与过拟合：分段验证

参数优化最容易翻车的地方不是算法，是评估方式。把 2020 到 2024 全区间拿来网格搜索，挑出夏普最高的那组，这叫在答案上训练。正确做法是切成三段：前段调参，中段验证，后段只跑一次看结果，看完不许回头改。

import itertools grid = dict(fast=[5, 10, 20, 30], slow=[40, 60, 90, 120], stop_loss=[0.05, 0.08, 0.12]) results = [] for combo in itertools.product(*grid.values()): p = dict(zip(grid.keys(), combo)) if p["fast"] >= p["slow"]: continue # 快线必须短于慢线，否则无意义 eng = build_engine(**p) r = eng.run(start="2020-01-01", end="2022-12-31") # 只在前段调 results.append({**p, "sharpe": r.sharpe, "mdd": r.max_drawdown, "turnover": r.turnover}) # 按夏普排序后取前 10 组，再拿中段单独验证，别在同一段上反复看 top10 = sorted(results, key=lambda x: -x["sharpe"])[:10]

itertools.product的搜索空间是笛卡尔积，三个参数各四档就是 64 组，加上成交价、手续费维度很快上千组。组数一多，夏普最高的那组几乎必然是噪声拟合出来的。经验值是：参数组数不要超过样本内交易笔数的十分之一，否则怎么调都能调出好看的结果。

还有一个细节：fast >= slow的组合要直接跳过，不然均线交叉的方向是反的，回测会跑出一堆无意义信号，白占时间还污染结果排序。

4.3 绩效指标：先看换手，再看夏普

看回测报告的顺序我改过好几次。现在的顺序是：换手率 → 最大回撤 → 收益分布 → 夏普 → 年化。换手率排第一，因为它最能暴露成本假设是否合理。年化 30% 配年换手 40 倍，扣掉双边成本和滑点基本归零，这种策略看着漂亮，实际上是在给券商打工。

最大回撤要看持续期，不只是幅度。回撤 15% 但三个月就修复，和回撤 15% 卡了两年，心理压力完全不是一个量级，后者在实盘里几乎必然被手动干预。教程里的绩效模块一般会输出回撤区间，翻一下那段时间发生了什么，比看数字有用。

夏普在样本少于两年时意义有限，尤其是交易笔数低于三十笔的策略，夏普高只能说明运气集中。我一般同时看盈亏比和胜率：胜率 35%、盈亏比 3 的策略，和胜率 65%、盈亏比 1 的策略，夏普可能差不多，但前者对连续亏损的容忍度要求高得多，仓位管理得跟着变。

5. 避坑排查：五条实际踩过的记录

5.1 回测年化四十，实盘两周回吐 —— 未来函数

现象：回测曲线近乎直线向上，月度胜率九成，实盘一上就亏，且亏损集中在开盘半小时。

原因：信号里用了当日收盘价，成交却按当日收盘价撮合。相当于收盘那一秒既知道结果又能成交，现实里这两件事差了一个交易日。还有一种变体是在计算均线时对整段历史做了center=True的滚动，等于用未来数据平滑当下。

解决：把所有成交价统一改成次日开盘价加滑点，信号只允许读hist.iloc[:-1]。改完之后年化掉到十几，那才是这个策略的真实水平。检查方法是对比“信号日”和“成交日”的日志时间戳，两者相同就一定有未来函数。

5.2 换个标的后收益归零 —— 复权错位

现象：同一个策略在 A 标的上年化二十，换到 B 标的上直接亏，但两个标的走势肉眼看着差不多。

原因：B 标的在回测区间内有过大比例送转，源数据是未复权价，除权日价格腰斩，策略当成暴跌触发止损。或者反过来，数据已经前复权，代码里又乘了一遍因子，早期价格被压到接近零。

解决：先打印除权日前后各五根 bar 的价格，看有没有不可能的单日跌幅。确认是复权问题后，统一在数据装载层做一次，别在策略里再做。前复权和后复权选一种，全程不混用——前复权适合看形态，后复权适合算收益。

5.3 日志里全是“下单成功”，持仓却不动 —— 取整与拒单

现象：策略日志显示发了几百笔订单，最终持仓始终是零，资金也没怎么变。

原因：手数取整后变成零。目标仓位 30%、资金十万、股价八十，算下来 375 股，向下取整到 300 股没问题；但如果资金只有一万，就是 37 股，取整到 0，订单直接作废。另一种情况是价格已经涨停，撮合模块拒单，但策略没检查返回值，日志里照样打“下单成功”。

解决：在下单函数里加一句if qty <= 0: log.warning(...)并统计，跑完看这条日志出现多少次。同时在撮合回调里记录拒单原因，涨跌停、成交量不足、资金不足分别计数，哪个占比高就修哪个。

5.4 多标的回测资金曲线翻倍 —— 资金共享与下单顺序

现象：单标的回测年化十，三个标的一起跑年化三十，且三个策略各自的收益加起来约等于组合收益，看起来像白捡了分散化的好处。

原因：每个策略各自按总资产的 30% 算目标仓位，三个加起来 90%，看似没超。但实际下单时没有任何共享约束，第一只标的用掉了 30% 现金，第二只仍然按原始总资产算仓位，第三只也是。资金超发，回测里凭空多出杠杆。

解决：把资金账户提到策略上层，所有下单先过统一的资金池，扣减可用资金后才发单。同一 bar 内多标的下单要指定顺序，我一般按流动性排序，先下成交概率高的，避免资金被不活跃标的占死。修完之后组合年化和单标的接近，那才是真实结果。

5.5 回测区间对不上 —— 时区与时间戳

现象：回测跑出来交易笔数比预期少一半，日志里开始日期和设定的差了八小时。

原因：行情数据带时区，策略里的start/end是不带时区的本地时间，比较时被隐式转换。日频数据看着没事，分钟频直接错位，某些 bar 被归到前一天，开收盘逻辑全乱。

解决：入库前统一转成不带时区的本地时间，或者全链路统一 UTC 并在展示层转换，但只能选一种。我习惯在数据装载函数的第一行就把时区剥掉，后面任何地方都不再出现tz_localize。跑完对比一下“实际首笔成交日”和“区间起始日”，差一天以上就回头查时区。

6. 进阶用法：多标的事件驱动与一套三遍验证法

单标的跑通之后，下一步是多标的。pyq 的多标模式不是在策略里循环 symbol，而是把同一时间戳的所有 bar 打包成一个事件一次性派发，策略拿到的是一个横截面。这个设计对截面策略友好，对时序策略要多写一层遍历，但资金共享天然成立——因为同一时刻只有一个策略在动钱。

切换时有两个地方要改。第一是on_bar的入参从单个 bar 变成 bar 列表，历史查询要按 symbol 分别取，取错了会拿到别的标的序列。第二是订单排序，同一时刻多个标的都要下单时，得先按可成交金额排序，否则资金先被流动性差的标的占住，好的机会反而错过。我一般按“近 20 日成交额降序”排，简单但有效。

验证环节我一直在用一个笨办法，叫三遍验证法：

第二遍是为了排除“任何信号在牛市里都能赚”的错觉。打乱信号后如果年化只掉三五个点，说明策略赚的是 beta，跟你的信号没关系。第三遍更狠，成本翻倍还能活的策略，实盘里才有容错空间；一翻倍就归零的，说明利润全靠成本假设撑着。

还有个小技巧：把回测区间随机抽十个子区间各跑一遍，看收益分布的离散程度。均值高、方差也大的策略，实盘里第一次遇到不利区间就可能被叫停，这种策略哪怕回测再好我也不会用。

说个自己的教训。早些年我调参数调到一个夏普 2.4 的组合，兴奋了一晚上，第二天把成交价从收盘改成次日开盘，夏普掉到 0.9。从那以后我每次开新策略，都强制先跑一遍“最差成交假设”——信号次日开盘成交、滑点拉满、成本翻倍，这三条都过了才允许看优化后的数字。慢是慢了点，但省下的都是实盘里的真金白银。希望帮到你。

本文还有配套的精品资源，点击获取