用Python复现通达信winner函数：手把手教你估算A股收盘获利比例（附完整代码）

用Python构建A股筹码分布模型：从零实现通达信winner函数

在量化投资领域，筹码分布分析是一个独特而实用的视角。不同于传统技术指标关注价格和成交量，筹码分布试图揭示不同价位上的持仓情况。这种分析方法最早出现在大智慧、通达信等专业股票软件中，其中winner函数就是计算收盘价下方筹码比例的核心工具。本文将带您从零开始，用Python完整复现这一功能，不仅理解其数学原理，还能将其应用于实际A股数据分析。

1. 理解筹码分布与winner函数

筹码分布理论的核心假设是：市场上的持仓成本会随着交易不断转移。当某日发生交易时，部分持仓会以新的价格成交，而剩余持仓则保持原有成本。winner函数的作用就是计算在当前收盘价之下，有多少比例的持仓处于盈利状态。

关键概念解析：

• 换手率：反映当日成交量占总流通股本的比例
• 平均成交价：当日成交金额除以成交量
• 筹码沉淀：未参与当日交易的持仓部分
• 获利比例：收盘价高于持仓成本的比例

筹码分布的计算遵循递推公式：

当日新筹码 = 总流通股本 × 当日换手率 沉淀筹码 = 前一日筹码 × (1 - 当日换手率)

2. 数据准备与预处理

要实现准确的筹码分布计算，我们需要获取以下核心数据：

import pandas as pd import numpy as np from datetime import datetime, timedelta def get_stock_data(stock_code, start_date, end_date): """ 获取股票交易数据 参数： stock_code: 股票代码，如'000001.SZ' start_date: 开始日期，'YYYY-MM-DD'格式 end_date: 结束日期，'YYYY-MM-DD'格式 返回： DataFrame包含columns: ['date', 'open', 'high', 'low', 'close', 'volume', 'amount'] """ # 这里应替换为实际数据接口调用 # 示例使用模拟数据 dates = pd.date_range(start_date, end_date) data = { 'date': dates, 'close': np.random.uniform(10, 20, len(dates)).cumsum(), 'volume': np.random.randint(100000, 500000, len(dates)), 'amount': np.random.uniform(1e6, 5e6, len(dates)) } return pd.DataFrame(data).set_index('date') def calculate_turnover_rate(volume, float_shares): """ 计算日换手率 参数： volume: 日成交量 float_shares: 流通股本 返回： 日换手率序列 """ return volume / float_shares

数据处理要点：

1. 确保数据包含成交量(volume)和成交额(amount)字段
2. 计算平均成交价：mean_price = amount / volume
3. 获取流通股本数据（可通过公开财报或专业数据接口）
4. 处理停牌日数据（成交量为零的日期）

3. 核心算法实现

基于筹码分布理论，我们构建winner函数的完整实现：

def calculate_chip_distribution(df, float_shares): """ 计算筹码分布 参数： df: 包含['volume', 'amount', 'close']的DataFrame float_shares: 流通股本 返回： 包含筹码分布结果的DataFrame """ df = df.copy() df['mean_price'] = df['amount'] / df['volume'] df['turnover'] = calculate_turnover_rate(df['volume'], float_shares) # 初始化筹码分布计算 chip_dist = pd.DataFrame(index=df.index) chip_dist[df.iloc[0]['mean_price']] = float_shares for i in range(1, len(df)): current_price = df.iloc[i]['mean_price'] turnover = df.iloc[i]['turnover'] # 计算新筹码 new_chip = float_shares * turnover # 更新已有筹码 prev_chip = chip_dist.iloc[i-1].copy() prev_chip *= (1 - turnover) # 合并筹码 current_dist = prev_chip.to_dict() if current_price in current_dist: current_dist[current_price] += new_chip else: current_dist[current_price] = new_chip # 保存结果 chip_dist = chip_dist.join(pd.DataFrame(current_dist, index=[df.index[i]])) return chip_dist.fillna(0) def winner_function(close_price, chip_distribution): """ 计算获利比例 参数： close_price: 当前收盘价 chip_distribution: 筹码分布Series 返回： 获利比例(0-1之间) """ profitable_chips = chip_distribution[chip_distribution.index <= close_price].sum() total_chips = chip_distribution.sum() return profitable_chips / total_chips if total_chips > 0 else 0

算法优化技巧：

1. 使用向量化操作提高计算效率
2. 对极值情况进行处理（如换手率>1的情况）
3. 添加平滑处理，避免价格微小波动导致结果突变
4. 实现滚动窗口计算，适应长期数据分析

4. 完整流程与验证测试

将各模块整合为完整工作流，并与专业软件结果对比：

def full_workflow(stock_code, start_date, end_date, float_shares): # 获取数据 df = get_stock_data(stock_code, start_date, end_date) # 计算筹码分布 chip_dist = calculate_chip_distribution(df, float_shares) # 计算每日获利比例 results = [] for date, row in chip_dist.iterrows(): close_price = df.loc[date, 'close'] winner_ratio = winner_function(close_price, row) results.append(winner_ratio) return pd.Series(results, index=chip_dist.index) # 示例使用 float_shares = 1e8 # 假设流通股本1亿股 results = full_workflow('000001.SZ', '2023-01-01', '2023-06-30', float_shares)

验证方法：

1. 边界测试：检查极端市场情况下的输出
2. 一致性测试：确保相同输入产生相同输出
3. 对比测试：与通达信软件结果进行比对
4. 敏感性分析：观察参数变化对结果的影响

常见差异原因：

• 流通股本数据的准确性
• 换手率计算方式的细微差别
• 停牌日处理逻辑不同
• 价格精度取舍差异

5. 高级应用与策略开发

掌握了winner函数的实现后，我们可以开发更复杂的量化策略：

策略示例：筹码集中度策略

def chip_concentration_strategy(stock_code, window=20, threshold=0.7): """ 筹码集中度策略 当近期winner值持续高于阈值时产生信号 """ data = full_workflow(stock_code, ...) signals = (data.rolling(window).mean() > threshold).astype(int).diff() return signals[signals > 0].index

扩展应用方向：

1. 结合其他技术指标构建复合策略
2. 开发筹码分布可视化工具
3. 研究不同市场周期下的筹码分布特征
4. 优化计算效率实现实时监控

性能优化建议：

• 使用numba加速核心计算
• 实现多进程并行计算
• 缓存中间结果减少重复计算
• 使用更高效的数据结构存储筹码分布

6. 模型局限性与改进方向

虽然筹码分布模型很有价值，但也存在一些固有局限：

主要局限性：

1. 假设所有交易均匀分布在当日价格区间
2. 无法区分主力资金和散户资金
3. 对极端行情（如连续涨停）适应性较差
4. 依赖准确的流通股本数据

改进方案：

def enhanced_winner_function(close_price, chip_distribution, volume_weight=True): """ 增强版winner函数 参数： volume_weight: 是否考虑成交量加权 """ if volume_weight: # 实现成交量加权逻辑 pass # 其他改进逻辑

实际应用中，我发现当结合5日均线过滤信号时，策略表现会明显提升。另外，在计算筹码分布时加入大宗交易数据修正，可以进一步提高准确性。