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

用Python构建股票获利比例分析工具：从筹码分布到winner函数全解析

在量化投资领域，准确评估当前市场参与者的盈亏状况是制定交易策略的关键一环。通达信软件中的winner函数正是这样一个实用工具，它能估算在特定价格水平下市场持仓的盈利比例。对于习惯使用Python进行数据分析的投资者来说，掌握如何用代码实现这一功能，意味着可以更灵活地将其整合到自定义策略中。本文将带你从零开始，理解筹码分布的核心逻辑，并用Python完整复现这一实用功能。

1. 理解winner函数的市场意义

winner函数的核心价值在于它反映了当前价格水平下市场参与者的盈亏结构。当一只股票的收盘价为15元时，winner函数会告诉我们有多少比例的持仓成本低于这个价格——这些持仓当前处于盈利状态。

为什么这个指标如此重要？从行为金融学角度看，持仓盈亏直接影响投资者的决策心理：

• 盈利持仓：投资者更倾向于获利了结，形成潜在卖压
• 亏损持仓：投资者可能选择"躺平"等待解套，减少短期抛售
• 盈亏平衡点：大量持仓成本集中在当前价格附近时，往往形成重要支撑或阻力

传统股票软件如通达信、大智慧都内置了这个函数，但作为Python使用者，我们需要理解其底层计算逻辑：

# 概念性代码展示 def conceptual_winner(price_distribution, current_price): """ 计算当前价格下盈利持仓比例 :param price_distribution: 各价格区间的持仓量分布 :param current_price: 当前市场价格 :return: 盈利比例[0,1] """ profitable = sum(amount for price, amount in price_distribution if price < current_price) total = sum(amount for _, amount in price_distribution) return profitable / total

2. 筹码分布建模的关键技术

由于无法获取每个投资者的真实持仓成本，我们需要通过市场公开数据来估算筹码分布。核心思路是利用历史成交数据和换手率信息，构建一个合理的概率模型。

2.1 换手率与持仓更新

换手率是连接不同交易日筹码分布的关键桥梁。假设某只股票总股本为1000万股：

对应的数学表达式为： 新价格持仓 = 总股本 × 当日换手率 旧价格持仓 = 前一日持仓 × (1 - 当日换手率)

2.2 Python实现筹码分布计算

我们需要用pandas高效处理这些递推关系。首先准备基础市场数据：

def prepare_market_data(context, stock_code, lookback_days=250): """ 获取计算所需的市场数据 :param context: 交易上下文，包含数据接口 :param stock_code: 股票代码 :param lookback_days: 回溯天数 :return: 包含成交量、成交额、换手率的DataFrame """ # 获取历史交易数据 df = context.get_market_data( ['volume', 'amount'], stock_code=stock_code, period='1d', count=lookback_days ) # 计算均价（元/股） df['mean_price'] = df['amount'] / df['volume'] / 100 # 获取换手率数据 turnover = context.get_turnover_rate( stock_code, start_date=df.index[0], end_date=df.index[-1] ) df['turnover'] = turnover[stock_code].values # 首日换手率特殊处理 df['turnover'].iloc[0] = 1.0 # 假设首日全部换手 return df[df['volume'] > 0] # 过滤无成交量交易日

3. 完整winner函数实现

现在我们将分布计算步骤整合成一个完整的解决方案。核心是计算每个历史价格对当前筹码分布的贡献权重。

3.1 核心计算逻辑

def calculate_profit_ratio(df, current_price): """ 计算在当前价格下的盈利比例 :param df: 包含筹码分布数据的DataFrame :param current_price: 当前市场价格 :return: 盈利比例[0,1] """ # 计算每个历史价格的留存权重 df['retention'] = (1 - df['turnover']).shift(1).fillna(1).cumprod() # 计算各价格对当前筹码的贡献 df['weight'] = df['turnover'] * df['retention'] # 筛选盈利持仓并计算比例 profitable = df[df['mean_price'] < current_price]['weight'].sum() total = df['weight'].sum() return profitable / total if total > 0 else 0.0

3.2 时序处理与批量计算

为了计算历史每一天的获利比例，我们需要实现滚动窗口处理：

def winner(context, close_prices, stock_code, lookback_days=250): """ 完整winner函数实现 :param context: 交易上下文 :param close_prices: 收盘价序列 :param stock_code: 股票代码 :param lookback_days: 回溯天数 :return: 各时点获利比例Series """ results = [] market_data = prepare_market_data(context, stock_code, lookback_days) for i in range(1, len(close_prices)+1): current_slice = close_prices[:i] current_price = current_slice[-1] # 获取与当前时点对应的市场数据 data_slice = market_data[market_data.index <= current_slice.index[-1]] if len(data_slice) < 2: # 至少需要2天数据 results.append(0.0) continue ratio = calculate_profit_ratio(data_slice, current_price) results.append(ratio) return pd.Series(results, index=close_prices.index)

4. 验证与优化策略

实现代码后，我们需要验证其合理性并探讨可能的优化方向。

4.1 结果验证方法

与专业软件对比时，注意以下几点：

1. 时间窗口一致性：确保回溯天数相同
2. 换手率计算口径：不同数据源可能有细微差异
3. 价格精度处理：四舍五入可能导致微小差别

典型验证代码结构：

def validate_results(py_ratio, software_ratio): """ 对比Python实现与专业软件结果 :param py_ratio: Python计算结果 :param software_ratio: 软件输出结果 :return: 差异分析报告 """ diff = py_ratio - software_ratio return { 'max_diff': diff.max(), 'mean_diff': diff.mean(), 'correlation': py_ratio.corr(software_ratio) }

4.2 性能优化技巧

当处理多只股票或长时间序列时，可以考虑：

• 向量化计算：用numpy替代部分pandas操作
• 并行处理：对多支股票使用multiprocessing
• 缓存机制：存储中间计算结果

优化后的权重计算示例：

def optimized_weight_calculation(turnover_series): """ 向量化计算权重 :param turnover_series: 换手率序列 :return: 权重数组 """ turnover = turnover_series.values retention = np.cumprod(np.concatenate([[1], 1 - turnover[:-1]])) return turnover * retention

5. 实际应用场景扩展

掌握了winner函数的核心实现后，我们可以将其融入更复杂的分析框架：

5.1 筹码峰识别

结合移动窗口统计，识别关键价格区域：

def identify_chip_peaks(price_distribution, window=5, threshold=0.1): """ 识别筹码密集区 :param price_distribution: 价格分布DataFrame :param window: 移动窗口大小 :param threshold: 峰值阈值 :return: 峰值价格列表 """ weights = price_distribution['weight'] rolling_mean = weights.rolling(window).mean() peaks = weights[(weights > rolling_mean) & (weights > threshold)] return price_distribution.loc[peaks.index, 'mean_price'].unique()

5.2 动态支撑阻力分析

将获利比例与成交量结合，判断潜在反转点：

def dynamic_support_resistance(profit_ratio, volume, ratio_thresholds=(0.2, 0.8)): """ 动态识别支撑阻力位 :param profit_ratio: 获利比例序列 :param volume: 成交量序列 :param ratio_thresholds: (支撑阈值, 阻力阈值) :return: 支撑/阻力信号Series """ signals = pd.Series(index=profit_ratio.index, dtype=str) low_cond = (profit_ratio < ratio_thresholds[0]) & (volume > volume.median()) signals[low_cond] = '支撑' high_cond = (profit_ratio > ratio_thresholds[1]) & (volume > volume.median()) signals[high_cond] = '阻力' return signals

在实盘环境中，建议先用小资金测试这些策略，同时注意市场环境变化对筹码分布模型的影响。不同板块、市值的股票可能需要调整参数设置，这需要结合历史回测数据不断优化。