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

用Python构建筹码分布估算器：从零实现winner函数核心逻辑

在量化交易和技术分析领域，理解市场参与者的持仓成本分布是制定策略的关键。传统股票软件中的winner函数正是这样一个工具，它通过估算不同价格区间的持仓比例，帮助投资者判断当前价格下盈利筹码的占比。本文将彻底拆解这个黑盒，用Python从零构建一个可解释、可扩展的筹码分布估算系统。

1. 筹码分布的本质与估算原理

筹码分布反映的是市场参与者在不同价格区间的持仓数量。由于无法获取真实持仓数据，我们只能通过成交量、成交额和换手率等公开信息进行估算。核心假设是：每日新增筹码的价格分布由当日成交均价决定，而历史筹码会随着换手率逐步衰减。

具体计算遵循三个基本原则：

1. 新筹码生成：当日成交量按成交均价形成新的筹码
2. 旧筹码衰减：历史筹码按当日换手率比例减少
3. 累积效应：多日数据叠加形成完整分布

以一个简单示例说明：

• 首日：1000万股以10元均价成交 → 筹码分布为{10元: 1000万}
• 次日：换手率20%，均价11元 →
  • 保留筹码：1000万×(1-20%)=800万@10元
  • 新增筹码：1000万×20%=200万@11元
• 第三日：换手率30%，均价12元 →
  • 10元筹码：800万×(1-30%)=560万
  • 11元筹码：200万×(1-30%)=140万
  • 新增筹码：1000万×30%=300万@12元

这种估算方法虽然简单，但能有效反映市场平均持仓成本的变化趋势。

2. 数据准备与预处理

完整的实现需要以下核心数据：

import pandas as pd import numpy as np # 模拟数据生成示例 def generate_mock_data(days=30): dates = pd.date_range(end=pd.Timestamp.today(), periods=days) base_volume = 1000000 # 基准成交量 base_price = 10.0 # 基准价格 data = { 'date': dates, 'volume': np.random.normal(base_volume, base_volume*0.2, days).astype(int), 'amount': np.random.normal(base_volume*base_price, base_volume*base_price*0.3, days), 'turnover': np.random.uniform(0.05, 0.5, days) # 换手率 } return pd.DataFrame(data).set_index('date') # 实际应用中可从tushare、akshare等接口获取真实数据 # df = pro.daily(ts_code='000001.SZ', start_date='20200101')

关键数据处理步骤：

1. 计算每日成交均价：

   df['mean_price'] = df['amount'] / df['volume']

2. 数据清洗：

   df = df[df['volume'] > 0] # 剔除零成交量数据 df['turnover'] = df['turnover'].clip(0.01, 0.99) # 控制换手率在合理范围

3. 时间排序：

   df = df.sort_index(ascending=True)

注意：实际应用中需考虑复权处理，建议使用后复权价格进行计算

3. 核心算法实现

我们采用面向对象的方式构建可复用的筹码分布计算器：

class ChipDistributionCalculator: def __init__(self, initial_volume=1e7): """ :param initial_volume: 初始流通股本(股) """ self.initial_volume = initial_volume self.chip_records = [] # 记录每日筹码分布 def _update_chips(self, current_chips, turnover, new_price): """ 更新筹码分布 :param current_chips: 当前筹码分布 {价格:数量} :param turnover: 当日换手率 :param new_price: 当日成交均价 :return: 更新后的筹码分布 """ updated_chips = {} # 旧筹码衰减 for price, amount in current_chips.items(): updated_chips[price] = amount * (1 - turnover) # 新增筹码 updated_chips[new_price] = updated_chips.get(new_price, 0) + self.initial_volume * turnover return updated_chips def compute_distribution(self, df): """ 计算完整历史筹码分布 :param df: 包含volume,amount,turnover的DataFrame """ current_chips = {} for date, row in df.iterrows(): mean_price = row['mean_price'] turnover = row['turnover'] if not current_chips: # 首日处理 current_chips = {mean_price: self.initial_volume} else: current_chips = self._update_chips(current_chips, turnover, mean_price) self.chip_records.append((date, dict(current_chips))) return self def calculate_winner(self, price): """ 计算指定价格下的获利比例 :param price: 目标价格 :return: 获利比例[0-1] """ if not self.chip_records: raise ValueError("请先调用compute_distribution方法") latest_chips = self.chip_records[-1][1] total = sum(latest_chips.values()) winning = sum(amt for p, amt in latest_chips.items() if p <= price) return winning / total if total > 0 else 0

使用示例：

# 初始化计算器 (假设初始流通股1亿股) calculator = ChipDistributionCalculator(initial_volume=1e8) # 计算历史筹码分布 calculator.compute_distribution(df) # 计算当前收盘价下的获利比例 latest_close = 12.5 win_ratio = calculator.calculate_winner(latest_close) print(f"在价格{latest_close}下的获利比例: {win_ratio:.2%}")

4. 验证与优化

4.1 结果验证方法

1. 交叉验证：与传统软件计算结果对比

   # 模拟验证数据 test_data = pd.DataFrame({ 'volume': [1e7]*5, 'amount': [10*1e7, 11*1e7, 12*1e7, 11*1e7, 10.5*1e7], 'turnover': [0, 0.2, 0.3, 0.25, 0.1] # 首日换手率为0 }) calculator = ChipDistributionCalculator(1e7) calculator.compute_distribution(test_data) assert abs(calculator.calculate_winner(11) - 0.72) < 0.01 # 预期值验证

2. 敏感性分析：观察换手率参数对结果的影响

4.2 性能优化技巧

对于长周期数据，原始实现可能较慢。以下是优化方案：

1. 向量化计算：

   def vectorized_update(df, initial_volume): df['remaining_ratio'] = (1 - df['turnover']).cumprod() df['new_chips'] = initial_volume * df['turnover'] * df['remaining_ratio'].shift(1, fill_value=1) return df.groupby('mean_price')['new_chips'].sum()

2. 并行计算：对多只股票计算时使用joblib并行

   from joblib import Parallel, delayed def batch_compute(stock_codes): return Parallel(n_jobs=4)( delayed(calculate_for_stock)(code) for code in stock_codes )

3. 缓存机制：存储中间结果避免重复计算

4.3 高级改进方向

1. 考虑成交量加权的价格分布：

   # 使用成交量加权计算更精细的价格分布 bins = np.linspace(df['mean_price'].min()*0.9, df['mean_price'].max()*1.1, 20) df['price_bin'] = pd.cut(df['mean_price'], bins=bins) weighted_dist = df.groupby('price_bin').apply( lambda x: (x['volume'] * x['turnover']).sum() )

2. 引入时间衰减因子：

   # 旧筹码随时间指数衰减 decay_factor = np.exp(-days_passed/halflife)

3. 结合订单簿数据：在有tick数据的情况下可构建更精确分布

5. 实际应用案例

将筹码分析整合到交易策略中：

class TradingStrategy: def __init__(self, chip_calculator): self.chip_calc = chip_calculator def generate_signal(self, price_data): """ 基于筹码分布生成交易信号 :return: 1(买入), -1(卖出), 0(持有) """ win_ratio = self.chip_calc.calculate_winner(price_data['close'][-1]) # 简单策略：当获利比例低于30%时买入，高于70%时卖出 if win_ratio < 0.3: return 1 elif win_ratio > 0.7: return -1 return 0 # 使用示例 strategy = TradingStrategy(calculator) daily_signal = strategy.generate_signal(price_df)

筹码分布与其他指标的结合分析：

指标组合	应用场景	判断逻辑
筹码分布 + RSI	超买超卖判断	低获利比例+高RSI→强烈卖出信号
筹码分布 + 成交量	支撑压力位确认	高筹码集中区+放量突破→有效突破信号
筹码分布 + MACD	趋势确认	筹码单峰分布+MACD金叉→趋势强化信号

典型分析流程：

1. 计算当前筹码分布及获利比例
2. 识别筹码密集区（支撑/压力位）
3. 结合价格相对位置判断突破可能性
4. 根据成交量验证突破有效性
5. 制定仓位管理策略

def analyze_chip_concentration(chip_dist): """分析筹码集中度""" prices = np.array(list(chip_dist.keys())) amounts = np.array(list(chip_dist.values())) mean = np.average(prices, weights=amounts) std = np.sqrt(np.average((prices-mean)**2, weights=amounts)) return { 'mean_price': mean, 'concentration': 1 - std/mean, 'peak_prices': find_peaks(amounts, prices) }

对于想要进一步探索的开发者，可以尝试将这套系统部署为实时分析服务，或者整合到backtrader、zipline等量化框架中。在实际使用中，我发现设置合理的换手率平滑窗口（如5日平均）能有效减少市场异常波动带来的噪声干扰。