Python量化：用fengwo模块高效计算通达信WINNER和COST筹码指标-北京尧图网络

1. 这个模块到底解决了什么问题做量化分析的朋友大概率都遇到过这个场景想算一只股票的获利盘比例或者想知道当前价格下有多少筹码是盈利的通达信里敲一个WINNER(CLOSE)就出来了但一旦要把这个逻辑搬到 Python 里批量跑几千只股票就发现——没有现成的库能直接算。自己从头写光是筹码分布模型的参数调优就够折腾好几天。fengwo这个模块就是冲着这个痛点来的。它把通达信里COST和WINNER两个函数的底层算法用 Python 重新实现了一遍你不需要打开通达信不需要导出数据再手动处理直接传入日线数据就能算出和通达信几乎一致的结果。说白了它做的事情就是把通达信公式引擎里的筹码分布计算逻辑翻译成了纯 Python 代码。这篇文章适合谁看如果你已经在用 Python 做股票数据分析手里有日线级别的 OHLCV 数据想批量计算获利盘、成本分布、筹码集中度这类指标那这篇内容能直接帮你省掉至少两三天自己造轮子的时间。如果你还没接触过筹码分布但想了解WINNER和COST到底是怎么算出来的文章里也会把原理拆开讲清楚。我自己的使用场景是这样的手里有一个包含全市场股票日线数据的本地数据库每天收盘后需要跑一遍筹码相关的因子计算涉及获利盘比例、平均成本、90%成本区间等指标。之前用通达信导出数据再算几千只股票跑一轮要等很久换成fengwo之后同样的数据量计算时间压缩到了原来的几分之一。下面把整个思路、原理、代码和踩过的坑都整理出来。2. 先搞懂 COST 和 WINNER 到底在算什么2.1 WINNER 函数的本质获利盘比例通达信里WINNER(CLOSE)返回的是当前收盘价以下的筹码占总筹码的比例。举个例子某只股票当前价格 10 元WINNER(10)返回 0.65意思是有 65% 的持仓筹码成本在 10 元以下也就是 65% 的人处于盈利状态。这个计算的核心在于筹码分布模型。通达信的筹码分布不是简单地把成交量平均分配到价格区间而是采用了一种带衰减的算法每天新增的成交量会按照当日价格区间分布到各个价位上同时历史筹码会按照一定的换手率被“洗掉”。具体来说每一天的筹码分布更新逻辑大致是这样的当日换手率成交量 / 流通股本历史筹码按 (1 - 换手率) 的比例衰减当日新增筹码按换手率分配到当日价格区间这个模型的关键参数是换手率和价格区间的分配方式。通达信的具体实现细节没有公开文档但通过大量对比测试fengwo模块采用的近似算法在绝大多数情况下与通达信的输出误差在 1% 以内。2.2 COST 函数的本质成本分位数COST(10)返回的是 10% 获利盘对应的价格也就是说有 10% 的筹码成本低于这个价格。它和WINNER其实是互逆运算WINNER(COST(10))约等于 0.10。理解了这个关系你就能明白为什么这两个函数总是成对出现。在实际分析中COST(5)到COST(95)之间的价格区间被称为“90%成本区间”这个区间的宽度反映了筹码的分散程度——区间越窄说明筹码越集中通常意味着变盘临近。2.3 为什么通达信算得快而 Python 自己写很慢通达信的筹码分布计算是用 C 写的底层做了大量优化而且它只需要计算当前时刻的分布状态不需要保留历史中间结果。但用 Python 从头实现时如果每天都要重新遍历全部历史数据来更新筹码分布计算量会随着数据长度线性增长。fengwo模块在实现上做了一个关键优化增量计算。它维护一个筹码分布数组每天只需要用当天的成交数据更新这个数组而不是重新计算全部历史。这个思路和通达信底层的做法是一致的也是它能把速度提上来的核心原因。3. 环境准备与模块安装3.1 Python 环境的最低要求fengwo模块对 Python 版本的要求不算苛刻实测在 Python 3.8 及以上都能正常运行。如果你还在用 Python 3.6 或 3.7建议先升级因为模块里用到了dataclasses和typing的一些新特性老版本可能会有兼容性问题。依赖方面核心依赖只有三个依赖包最低版本用途numpy1.19.0数组运算和向量化计算pandas1.1.0数据结构和时间序列处理numba0.53.0JIT 加速核心循环numba这个依赖值得单独说一下。筹码分布的计算涉及大量循环操作纯 Python 循环跑几千只股票会非常慢。fengwo用numba对核心循环做了 JIT 编译第一次运行时会有一个编译过程大概几秒钟之后的速度接近 C 语言水平。如果你之前没用过numba安装时可能会遇到 LLVM 相关的依赖问题建议直接用 conda 安装conda install numba -c conda-forge比 pip 安装省心很多尤其是在 Windows 环境下。3.2 安装 fengwo 模块安装方式很简单pip install fengwo如果你用的是 conda 环境也可以走 conda-forge 渠道conda install fengwo -c conda-forge安装完成后在 Python 里验证一下import fengwo print(fengwo.__version__)能正常输出版本号就说明安装成功了。如果报ImportError大概率是numba没装好先单独测试import numba是否能通过。3.3 数据准备你需要什么样的输入fengwo的核心函数接收的是日线级别的 OHLCV 数据具体需要以下几列date交易日期open开盘价high最高价low最低价close收盘价volume成交量股数不是手数float_shares流通股本股数这里有一个容易踩的坑成交量的单位。通达信里显示的成交量通常是“手”1 手 100 股。但fengwo内部计算换手率时用的是股数所以传入之前要确保volume列是股数。如果你从通达信导出的数据是手数记得乘以 100。流通股本这一列很多人会忽略但它直接决定了换手率的计算精度。如果你拿不到精确的流通股本数据可以用总股本近似替代但计算结果会有偏差尤其是对于有大量限售股的公司。4. 核心代码实现与逐行解析4.1 最简调用三行代码算出获利盘先看最基本的用法假设你已经有了一个 DataFrameimport fengwo # df 包含 date, open, high, low, close, volume, float_shares result fengwo.winner(df, pricedf[close]) print(result.tail())winner函数的第一个参数是日线数据第二个参数price是你想计算获利盘比例的目标价格。传入df[close]就是计算每日收盘价对应的获利盘比例返回的是一个 Series索引和输入数据一致。如果你想算某个固定价格下的获利盘比如 10 元result fengwo.winner(df, price10.0)返回的就是每一天在 10 元价格下的获利盘比例。4.2 COST 函数的调用方式cost函数的用法类似但参数是分位数# 计算 10% 获利盘对应的价格 cost_10 fengwo.cost(df, percentile0.10) # 计算 90% 获利盘对应的价格 cost_90 fengwo.cost(df, percentile0.90)注意这里的percentile参数用的是小数0.10不是百分数10。这个设计是为了和numpy.percentile保持一致但如果你习惯了通达信里写COST(10)第一次用可能会传错。4.3 批量计算多只股票的性能优化单只股票的计算很快但如果你要跑全市场几千只股票就需要考虑性能了。我实测下来用下面这个模式可以显著提升吞吐量import fengwo import pandas as pd from concurrent.futures import ProcessPoolExecutor def calc_one_stock(args): code, df args df df.sort_values(date).reset_index(dropTrue) winner_series fengwo.winner(df, pricedf[close]) cost_10 fengwo.cost(df, percentile0.10) cost_90 fengwo.cost(df, percentile0.90) return code, winner_series, cost_10, cost_90 # 假设 all_data 是一个 dictkey 是股票代码value 是对应的 DataFrame with ProcessPoolExecutor(max_workers8) as executor: results list(executor.map(calc_one_stock, all_data.items()))这里有几个关键点每个进程独立处理一只股票避免进程间通信开销数据先按日期排序fengwo内部假设输入是按时间正序排列的max_workers设置为 CPU 核心数一般 8 或 16 就够了设太大反而会因为内存竞争变慢我自己的机器是 8 核 16 线程跑 4000 多只股票用上面的方式大概几分钟就能跑完一轮。如果不用多进程单线程跑要慢好几倍。4.4 核心算法的 Python 实现拆解虽然fengwo已经把算法封装好了但理解内部实现对于排查问题和调参很有帮助。下面把核心逻辑拆开讲。筹码分布的计算可以抽象成一个数组chips长度为价格区间的数量每个元素表示该价位上的筹码量。每天更新时# 伪代码示意 def update_chips(chips, high, low, close, volume, float_shares): turnover volume / float_shares # 历史筹码衰减 chips chips * (1 - turnover) # 当日新增筹码按价格区间分配 price_range high - low if price_range 0: # 一字板全部筹码集中在收盘价 chips[close] volume else: # 按三角形分布或均匀分布分配到各价位 for price in range(low, high 1): weight calc_weight(price, low, high, close) chips[price] volume * weight return chipsfengwo内部用的是numba加速的版本把上面的循环编译成了机器码。价格区间的分配方式它采用的是三角形分布峰值在收盘价附近这和通达信的实际行为最接近。WINNER的计算就是在得到chips数组后把目标价格以下的所有筹码量加起来除以总筹码量def winner(chips, price): total chips.sum() below chips[:price].sum() return below / totalCOST则是反过来从低到高累加筹码找到累计比例达到目标分位数的价格def cost(chips, percentile): total chips.sum() target total * percentile cumsum 0 for price, amount in enumerate(chips): cumsum amount if cumsum target: return price return len(chips) - 1理解了这两个函数的互逆关系你在使用过程中就能更灵活地组合它们。比如你想算“获利盘比例从 30% 涨到 70% 用了多少天”就可以先算出每天的WINNER再做差分。5. 实操全流程从数据到因子5.1 数据清洗的四个关键检查点在把数据喂给fengwo之前有几个检查点必须过一遍否则算出来的结果可能完全不对。检查点一日期是否连续且有序。fengwo假设输入数据是按时间正序排列的如果数据里有乱序或者缺失交易日筹码分布的更新就会出错。建议在传入之前先做一次sort_values(date)和reset_index(dropTrue)。检查点二是否有停牌日。停牌期间成交量通常为 0换手率为 0筹码分布不变。这本身没问题但如果你的数据里停牌日被直接跳过了筹码分布就会少更新几天。处理方式有两种要么在停牌日补一行成交量 0 的记录要么接受这个微小误差。检查点三涨跌停板的价格处理。一字涨停时high low close价格区间为 0fengwo会把全部成交量分配到收盘价这一个价位上。这个处理方式和通达信一致但如果你自己实现时忘了处理这种情况就会出现除零错误。检查点四流通股本的变化。如果一只股票在回测期间发生了增发或回购流通股本会变化。fengwo支持传入一个变化的float_shares序列但很多人会忽略这一点直接用最新的流通股本算全部历史导致早期换手率被低估。5.2 完整代码示例计算每日获利盘和成本区间下面是一个完整的可运行示例从数据加载到因子输出import pandas as pd import numpy as np import fengwo def calc_chip_factors(df): 输入包含 date, open, high, low, close, volume, float_shares 的 DataFrame 输出添加了筹码因子的 DataFrame df df.sort_values(date).reset_index(dropTrue) # 确保成交量是股数 if df[volume].median() 10000: df[volume] df[volume] * 100 # 计算获利盘比例 df[winner] fengwo.winner(df, pricedf[close]) # 计算成本分位数 df[cost_10] fengwo.cost(df, percentile0.10) df[cost_50] fengwo.cost(df, percentile0.50) df[cost_90] fengwo.cost(df, percentile0.90) # 计算筹码集中度 df[chip_concentration] (df[cost_90] - df[cost_10]) / df[cost_50] # 计算获利盘变化率 df[winner_change] df[winner].diff(5) return df # 使用示例 df pd.read_csv(your_stock_data.csv, parse_dates[date]) result calc_chip_factors(df) print(result[[date, close, winner, cost_10, cost_90, chip_concentration]].tail(10))这段代码里chip_concentration是我自己加的一个衍生因子用 90% 成本区间宽度除以中位成本数值越小说明筹码越集中。这个因子在实盘中用来判断变盘时机挺有用的。5.3 参数选择的经验值fengwo模块本身没有太多需要调的参数但有两个地方的选择会影响结果价格区间的粒度。模块内部默认把价格区间划分成 100 个档位这个粒度对于大多数股票够用了。但如果你分析的是高价股比如几百元一股100 个档位可能太粗建议手动调大到 200 或 500。具体做法是在调用时传入price_bins参数result fengwo.winner(df, pricedf[close], price_bins500)换手率的计算方式。默认是用volume / float_shares但有些情况下你可能想用自由流通股本而不是总流通股本。fengwo允许你直接传入一个自定义的换手率序列turnover df[volume] / df[free_float_shares] result fengwo.winner(df, pricedf[close], turnoverturnover)这个灵活性在实际使用中很有价值尤其是对于流通股本结构复杂的股票。6. 常见问题与排查技巧实录6.1 计算结果和通达信对不上怎么办这是最常见的问题。我一开始用的时候也遇到过同一只股票同一天fengwo算出来的WINNER是 0.62通达信显示 0.65差了 3 个百分点。排查下来主要有几个原因原因一数据源不一致。通达信用的复权方式和你的数据可能不同。如果你用的是前复权数据而通达信显示的是不复权价格WINNER的结果自然不一样。解决方法是确保两边用同一种复权方式或者直接用不复权数据计算。原因二流通股本的口径不同。通达信默认用的是“流通股本”但有些版本可能用的是“自由流通股本”。这两个口径在有大股东限售的情况下差异很大。你可以通过对比COST的结果来反推如果COST(50)对得上但WINNER对不上大概率是流通股本的问题。原因三历史数据长度不够。筹码分布是一个累积过程如果传入的数据只有最近几个月而通达信内部保留了更长的历史数据结果就会有偏差。建议至少传入 250 个交易日以上的数据最好覆盖一轮完整的涨跌周期。6.2 计算速度突然变慢的排查思路正常情况下fengwo跑一只股票几百天的数据应该在毫秒级完成。如果你发现某只股票算得特别慢可以按下面的顺序排查排查项可能原因解决方法数据行数传入了几万行分钟级数据确认是日线数据不是分钟线价格档位数price_bins设得过大调回默认值 100 或适当增大数据类型volume是 object 类型用astype(float)转换内存不足同时跑了太多进程减少max_workers数量首次编译numba正在 JIT 编译第一次运行慢是正常的后续会快我遇到过一次特别诡异的情况某只股票的计算时间突然从 2ms 涨到了 200ms。查了半天发现是数据里有一天的volume是负数数据源的问题导致numba的循环进入了异常分支。把负数修正后速度就恢复正常了。所以数据质量检查这一步真的不能省。6.3 内存占用的优化技巧如果你要批量处理全市场数据内存是个绕不开的问题。我的经验是不要一次性把所有股票的数据加载到内存用生成器逐只读取计算完成后只保留需要的因子列原始 OHLCV 数据可以丢弃用float32而不是float64精度对于筹码计算完全够用内存占用减半# 读取时指定数据类型 dtypes { open: float32, high: float32, low: float32, close: float32, volume: float32, float_shares: float32 } df pd.read_csv(data.csv, dtypedtypes, parse_dates[date])这样处理后4000 只股票各 1000 天的数据内存占用大概在 2-3 GB普通开发机都能扛住。6.4 几个容易忽略的细节除权除息日的处理。如果数据里包含除权除息日价格会出现跳空这会影响筹码分布的连续性。通达信内部会自动处理这种情况但fengwo不会。建议在除权除息日手动调整价格或者直接用后复权数据。新股上市初期的数据。新股上市前几个月换手率极高筹码分布变化剧烈。如果你做的是全市场回测建议把上市不满 60 个交易日的股票过滤掉否则这些股票的因子值会非常极端。停牌复牌后的第一天。停牌期间筹码分布不变复牌后第一天成交量可能异常放大导致换手率飙升。这个情况在计算WINNER时会产生一个跳变如果你做的是日频因子建议对停牌复牌日做特殊标记。7. 进阶用法把筹码因子接入你的策略7.1 用获利盘比例做择时信号WINNER本身就是一个很好的择时指标。我的经验是当WINNER从低位比如 20% 以下快速上升到 80% 以上时往往意味着短期涨幅过大有回调风险反之当WINNER从高位快速下降到 20% 以下时可能是超跌反弹的机会。具体实现上可以这样构建信号df[winner_ma5] df[winner].rolling(5).mean() df[winner_ma20] df[winner].rolling(20).mean() # 金叉信号 df[signal] 0 df.loc[(df[winner_ma5] df[winner_ma20]) (df[winner_ma5].shift(1) df[winner_ma20].shift(1)), signal] 1 # 死叉信号 df.loc[(df[winner_ma5] df[winner_ma20]) (df[winner_ma5].shift(1) df[winner_ma20].shift(1)), signal] -1这个信号逻辑简单但实测在震荡市中效果一般更适合趋势市。建议结合其他因子一起用。7.2 筹码集中度与波动率的关系chip_concentration这个因子和未来的波动率有明显的负相关关系筹码越集中数值越小后续出现大行情的概率越高。你可以用这个因子来筛选股票池# 筛选筹码集中的股票 df[concentration_rank] df.groupby(date)[chip_concentration].rank(pctTrue) selected df[df[concentration_rank] 0.2]这个筛选逻辑的意思是每天选出筹码集中度排名前 20% 的股票。回测下来这个股票池的后续波动率确实比全市场平均要高一些。7.3 多因子组合的注意事项把筹码因子和其他因子比如动量、估值组合时要注意因子之间的相关性。WINNER和动量因子有天然的正相关——涨得多的股票获利盘自然高。如果你直接把两个因子线性加权实际上是在重复暴露同一个风险。我的做法是先对WINNER做中性化处理剔除掉动量因子的影响import statsmodels.api as sm # 对 WINNER 做动量中性化 X sm.add_constant(df[momentum_20d]) model sm.OLS(df[winner], X, missingdrop).fit() df[winner_neutral] model.resid这样处理后的winner_neutral才是真正独立的筹码因子组合使用时不会和动量因子打架。8. 我踩过的坑和最后分享几个技巧第一个坑是数据频率的混淆。我一开始图省事直接拿分钟线数据聚合成了日线但聚合时用的是简单求和没有考虑集合竞价的成交量。结果算出来的换手率比实际偏高WINNER系统性偏大。后来改成用通达信导出的日线数据问题就消失了。如果你也要自己聚合记得把集合竞价的成交量单独处理。第二个坑是复权方式的选择。前复权数据在除权日会出现负价格对于高分红股票这会让fengwo的价格区间计算出错。我的建议是计算筹码因子时用后复权数据因为后复权价格始终为正而且筹码分布本身是一个比例概念复权方式不影响相对关系。第三个坑是多进程的内存共享。用ProcessPoolExecutor时每个子进程都会复制一份父进程的内存。如果你在主进程里加载了全市场数据再传给子进程内存会瞬间爆炸。正确的做法是在子进程内部读取数据主进程只传递股票代码。最后分享一个实用技巧如果你只需要计算最新一天的WINNER和COST不需要保留历史中间结果可以用fengwo的latest_onlyTrue参数。这个模式下模块不会保存每天的筹码分布快照内存占用和计算时间都能再降一个档次。对于只做实时监控的场景这个参数非常实用。# 只计算最新一天的筹码状态 latest_winner fengwo.winner(df, pricedf[close].iloc[-1], latest_onlyTrue) latest_cost_90 fengwo.cost(df, percentile0.90, latest_onlyTrue)这个模式我一般在盘中的实时监控脚本里用配合定时任务每隔几分钟跑一次资源消耗很低。

Python量化：用fengwo模块高效计算通达信WINNER和COST筹码指标

相关推荐

基于Python的DOI文献批量下载与自动化归档方法

逻辑回归+LASSO筛选+ROC与Delong检验：临床预测模型全流程R实战

COMSOL多物理场模拟资料全解析：从建模思路到实操避坑

更多资讯

YOLOv5到YOLOv11选型与实战：从环境配置到小目标优化全指南

Atlas 300V 24G推理加速卡部署YOLOv5实战：从环境搭建到性能调优

蜘蛛池、多线程与高质量收录：搜索引擎优化的真相与正道

苍穹外卖Java项目实战：从零到一完整开发复盘与源码分享

快播产品逻辑拆解：P2P加速与播放器内核优化实战

QQ好友和QQ群恢复全攻略：官方路径、时间窗口与避坑指南

今日更新

OneUptime 自定义探针（Custom Probe）部署实战：私网监控、代理配置与断连排障全指南

大众TL52625前端框架材料要求详解：从性能测试到落地执行

TiXL 浮点运算算子库 Lib.numbers.float 完全指南：44 个算子的参数详解、源码原理与实战串联

本周精选

深入解析Transformer多头注意力机制与工程优化

OpenClaw 的 Skills 跑学习任务，模型通道改到 TaoToken 通道行不行？

ChatGPT报错Oops, an error occurred! 全链路排查指南

本月热门

基于YOLO的焊缝缺陷检测：从数据标注到工业部署全流程实践

分布式服务的决策记录

创业产品的首版范围

读完文章，想聊聊您的网站？