Python实现筹码分布核心指标WINNER,附完整代码

import pandas as pdimport numpy as npdef winner_simple(close_prices, current_price, window=250):    """    最简版WINNER函数 - 滑窗比例法    参数:        close_prices: 历史收盘价序列        current_price: 当前价格        window: 统计窗口期（默认250个交易日）    返回:        获利盘比例 (0-1)    """    # 取最近window天的数据    recent_prices = close_prices[-window:]    # 计算低于当前价格的比例    winner_ratio = (recent_prices <= current_price).sum() / len(recent_prices)    return winner_ratio# 使用示例# 假设df是包含收盘价的数据框df['winner_simple'] = df['close'].rolling(250).apply(    lambda x: (x <= x.iloc[-1]).sum() / len(x))

import pandas as pdimport numpy as npfrom typing import Tupleclass ChipAnalysis:    """筹码分布分析类"""    def __init__(self, df: pd.DataFrame, decay_coeff: float = 1.0):        """        初始化        参数:            df: 包含open, high, low, close, volume的DataFrame            decay_coeff: 换手衰减系数，默认1.0        """        self.df = df.copy()        self.decay_coeff = decay_coeff        self.chip_distribution = None  # 存储筹码分布    def distribute_volume(self, row: pd.Series, n_bins: int = 100) -> np.ndarray:        """        将单日成交量分配到价格区间（三角分布模型）        参数:            row: 单日数据（含high, low, close, volume）            n_bins: 划分的格子数（精度）        返回:            各价格区间的筹码分配数组        """        high, low, close, volume = row['high'], row['low'], row['close'], row['volume']        # 生成价格格子        price_bins = np.linspace(low, high, n_bins + 1)        bin_width = (high - low) / n_bins        # 三角分布: 在close处密度最高，向high和low线性递减        # 计算每个格子的概率密度        densities = np.zeros(n_bins)        for i in range(n_bins):            bin_center = (price_bins[i] + price_bins[i+1]) / 2            if bin_center <= close:                # 左半边：线性增加                density = (bin_center - low) / (close - low) if close > low else 1            else:                # 右半边：线性减少                density = (high - bin_center) / (high - close) if high > close else 1            densities[i] = max(0, min(1, density))        # 归一化并乘以成交量        densities = densities / densities.sum() if densities.sum() > 0 else densities        return densities * volume * self.decay_coeff    def build_chip_distribution(self, lookback: int = 250, n_bins: int = 100):        """        构建筹码分布模型        参数:            lookback: 回溯周期（天）            n_bins: 价格区间划分精度        """        # 初始化筹码分布数组        price_range = self.df['high'].max() - self.df['low'].min()        min_price = self.df['low'].min() - price_range * 0.1        max_price = self.df['high'].max() + price_range * 0.1        price_edges = np.linspace(min_price, max_price, n_bins + 1)        # 累积筹码        total_chips = np.zeros(n_bins)        for idx, row in self.df.tail(lookback).iterrows():            daily_chips = self.distribute_volume(row, n_bins)            # 应用衰减（可选）            total_chips = total_chips * 0.998 + daily_chips        self.chip_distribution = total_chips        self.price_edges = price_edges        self.bin_centers = (price_edges[:-1] + price_edges[1:]) / 2    def winner(self, price: float) -> float:        """        计算获利盘比例 WINNER(price)        参数:            price: 目标价格（如当前收盘价）        返回:            获利盘比例 (0-1)        """        if self.chip_distribution is None:            raise ValueError("请先调用 build_chip_distribution() 构建筹码分布")        # 找到价格所在的区间索引        idx = np.searchsorted(self.bin_centers, price)        # 累计获利筹码        winner_chips = self.chip_distribution[:idx].sum()        total_chips = self.chip_distribution.sum()        return winner_chips / total_chips if total_chips > 0 else 0    def cost(self, percentile: float) -> float:        """        计算成本分布百分位 COST(percentile)        参数:            percentile: 百分位 (0-100)        返回:            对应百分位的价格        """        if self.chip_distribution is None:            raise ValueError("请先调用 build_chip_distribution() 构建筹码分布")        target_volume = self.chip_distribution.sum() * percentile / 100        cumsum = np.cumsum(self.chip_distribution)        idx = np.searchsorted(cumsum, target_volume)        idx = min(idx, len(self.bin_centers) - 1)        return self.bin_centers[idx]# 使用示例# 准备数据（需包含OHLCV）# df = yf.download('000001.SS', period='1y')# 创建分析器# chip = ChipAnalysis(df)# chip.build_chip_distribution(lookback=250)# 计算当前获利盘比例# winner_value = chip.winner(df['close'].iloc[-1])# print(f"获利盘比例: {winner_value:.2%}")# 计算90%成本分布# cost_90 = chip.cost(90)# print(f"90%成本线: {cost_90:.2f}")

pip install fengwo

import fengwoimport pandas as pdimport yfinance as yf# 获取数据df = yf.download('AAPL', period='6mo', interval='1d')# 方法1：直接计算WINNERwinner_values = fengwo.WINNER(df['Close'].values)# 方法2：计算成本分布cost_50 = fengwo.COST(df, 50)  # 50%成本线print(f"最新获利盘比例: {winner_values[-1]:.2%}")

from concurrent.futures import ThreadPoolExecutorimport fengwodef calc_winner(symbol):    df = yf.download(symbol, period='6mo')    return fengwo.WINNER(df['Close'].values)[-1]symbols = ['AAPL', 'GOOGL', 'MSFT', 'TSLA']with ThreadPoolExecutor(max_workers=4) as executor:    results = dict(zip(symbols, executor.map(calc_winner, symbols)))for symbol, winner in results.items():    print(f"{symbol}: 获利盘比例 {winner:.2%}")

def winner_pick_stocks(df_dict, winner_threshold_low=0.1, winner_threshold_high=0.9):    """    基于WINNER的选股策略    参数:        df_dict: {股票代码: DataFrame} 字典        winner_threshold_low: 底部阈值（超卖）        winner_threshold_high: 顶部阈值（超买）    返回:        信号字典    """    signals = {}    for symbol, df in df_dict.items():        # 计算WINNER        chip = ChipAnalysis(df)        chip.build_chip_distribution(lookback=250)        winner = chip.winner(df['close'].iloc[-1])        # 信号判断        if winner < winner_threshold_low:            signals[symbol] = {'signal': 'BUY', 'winner': winner, 'reason': '底部区域'}        elif winner > winner_threshold_high:            signals[symbol] = {'signal': 'SELL', 'winner': winner, 'reason': '顶部区域'}        else:            signals[symbol] = {'signal': 'HOLD', 'winner': winner, 'reason': '中性区域'}    return signals# 实战示例# stock_data = {}# for symbol in ['000001.SS', '000002.SZ', '600036.SS']:#     stock_data[symbol] = yf.download(symbol, period='1y')#     # results = winner_pick_stocks(stock_data)# for symbol, info in results.items():#     print(f"{symbol}: {info['signal']} ({info['reason']}) - 获利盘:{info['winner']:.2%}")