import akshare as akimport pandas as pdimport numpy as npfrom datetime import datetime, timedeltaclass XueqiuHotStockAnalyzer:    def __init__(self):        self.stock_data = {}    def get_hot_stock_list(self):        """获取雪球热股榜单"""        try:            df = ak.stock_xq_hot_rank()            return df        except Exception as e:            print(f"获取数据失败: {e}")            return None    def calculate_hotness_factor(self, df, lookback_days=5):        """计算热度因子"""        if df is None or len(df) == 0:            return None        # 基础热度因子（基于关注人数）        df['base_hotness'] = df['关注人数'] / df['关注人数'].max()        # 排名稳定性因子（排名变化越小，稳定性越高）        if '排名' in df.columns:            df['rank_stability'] = 1 / (df['排名'].diff().abs() + 1)        # 关注度增长率因子        # 这里需要历史数据，假设我们有历史关注度数据        # df['attention_growth'] = df['关注人数'].pct_change(periods=lookback_days)        # 综合热度因子（加权平均）        weights = {'base_hotness': 0.4, 'rank_stability': 0.3}        df['comprehensive_hotness'] = sum(df[factor] * weight                                          for factor, weight in weights.items())        return df.sort_values('comprehensive_hotness', ascending=False)    def analyze_sector_distribution(self, df, top_n=50):        """分析行业分布"""        if df is None or len(df) == 0:            return None        sector_analysis = df.head(top_n).groupby('所属行业').agg({            '股票代码': 'count',            '关注人数': 'sum',            '最新价': 'mean'        }).rename(columns={'股票代码': '股票数量'})        sector_analysis['行业占比'] = sector_analysis['股票数量'] / top_n        return sector_analysis.sort_values('股票数量', ascending=False)

class HotStockMomentumStrategy:    def __init__(self, initial_capital=100000):        self.initial_capital = initial_capital        self.portfolio = {}    def generate_signals(self, hot_stocks_df, price_data,                         hotness_threshold=0.7,                         min_price_change=0.05):        """        生成交易信号        hotness_threshold: 热度阈值（0-1）        min_price_change: 最小价格变化要求        """        signals = []        for _, stock in hot_stocks_df.iterrows():            stock_code = stock['股票代码']            stock_name = stock['股票名称']            hotness = stock.get('comprehensive_hotness', 0)            # 获取价格数据            if stock_code in price_data:                current_price = price_data[stock_code]['current']                prev_price = price_data[stock_code]['prev']                price_change = (current_price - prev_price) / prev_price                # 买入信号：热度高且价格上涨                if (hotness >= hotness_threshold and                     price_change >= min_price_change):                    signals.append({                        '股票代码': stock_code,                        '股票名称': stock_name,                        '信号类型': '买入',                        '热度分数': hotness,                        '价格涨幅': price_change,                        '建议仓位': min(0.1, 0.2 * hotness)  # 仓位控制                    })                # 卖出信号：热度下降且价格下跌                elif (hotness < hotness_threshold * 0.7 and                       price_change < -min_price_change):                    signals.append({                        '股票代码': stock_code,                        '股票名称': stock_name,                        '信号类型': '卖出',                        '热度分数': hotness,                        '价格涨幅': price_change                    })        return pd.DataFrame(signals)    def backtest_strategy(self, signals_df, price_history,                          commission_rate=0.0003):        """策略回测"""        capital = self.initial_capital        positions = {}        trade_history = []        for _, signal in signals_df.iterrows():            stock_code = signal['股票代码']            signal_type = signal['信号类型']            if signal_type == '买入':                # 计算买入数量                position_value = capital * signal['建议仓位']                buy_price = price_history[stock_code]['buy_price']                quantity = int(position_value / buy_price)                # 计算成本                cost = quantity * buy_price                commission = cost * commission_rate                total_cost = cost + commission                if total_cost <= capital:                    capital -= total_cost                    positions[stock_code] = {                        'quantity': quantity,                        'buy_price': buy_price,                        'buy_date': datetime.now()                    }                    trade_history.append({                        '日期': datetime.now(),                        '操作': '买入',                        '股票': stock_code,                        '数量': quantity,                        '价格': buy_price,                        '成本': total_cost                    })            elif signal_type == '卖出' and stock_code in positions:                position = positions[stock_code]                sell_price = price_history[stock_code]['sell_price']                # 计算收益                revenue = position['quantity'] * sell_price                commission = revenue * commission_rate                net_revenue = revenue - commission                capital += net_revenue                # 计算收益率                buy_cost = position['quantity'] * position['buy_price']                return_rate = (net_revenue - buy_cost) / buy_cost                trade_history.append({                    '日期': datetime.now(),                    '操作': '卖出',                    '股票': stock_code,                    '数量': position['quantity'],                    '价格': sell_price,                    '收益': net_revenue - buy_cost,                    '收益率': return_rate                })                del positions[stock_code]        # 计算最终绩效        final_value = capital + sum(            pos['quantity'] * price_history[code]['current']             for code, pos in positions.items()        )        total_return = (final_value - self.initial_capital) / self.initial_capital        return {            '初始资金': self.initial_capital,            '最终价值': final_value,            '总收益率': total_return,            '交易次数': len(trade_history),            '持仓数量': len(positions),            '交易记录': pd.DataFrame(trade_history)        }

import matplotlib.pyplot as pltimport seaborn as snsfrom matplotlib import font_manager# 设置中文字体plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = Falseclass XueqiuVisualizer:    def __init__(self):        self.figsize = (15, 10)    def plot_hot_stock_ranking(self, df, top_n=20):        """绘制热股排名图"""        fig, axes = plt.subplots(2, 2, figsize=self.figsize)        # 1. 关注人数TOP N        top_stocks = df.nlargest(top_n, '关注人数')        axes[0, 0].barh(top_stocks['股票名称'], top_stocks['关注人数'])        axes[0, 0].set_title(f'关注人数TOP {top_n}')        axes[0, 0].set_xlabel('关注人数（万）')        # 2. 热度分数分布        axes[0, 1].hist(df['comprehensive_hotness'], bins=20, alpha=0.7)        axes[0, 1].set_title('热度分数分布')        axes[0, 1].set_xlabel('热度分数')        axes[0, 1].set_ylabel('股票数量')        # 3. 行业分布        sector_dist = df['所属行业'].value_counts().head(10)        axes[1, 0].pie(sector_dist.values, labels=sector_dist.index, autopct='%1.1f%%')        axes[1, 0].set_title('热门股票行业分布')        # 4. 价格与热度散点图        axes[1, 1].scatter(df['最新价'], df['comprehensive_hotness'], alpha=0.5)        axes[1, 1].set_title('价格 vs 热度')        axes[1, 1].set_xlabel('最新价（元）')        axes[1, 1].set_ylabel('热度分数')        plt.tight_layout()        plt.show()    def plot_strategy_performance(self, backtest_results):        """绘制策略绩效图"""        fig, axes = plt.subplots(2, 2, figsize=self.figsize)        # 1. 资金曲线        trade_history = backtest_results['交易记录']        if len(trade_history) > 0:            capital_curve = [backtest_results['初始资金']]            current_capital = backtest_results['初始资金']            for _, trade in trade_history.iterrows():                if trade['操作'] == '买入':                    current_capital -= trade['成本']                else:                    current_capital += trade['收益']                capital_curve.append(current_capital)            axes[0, 0].plot(capital_curve)            axes[0, 0].set_title('资金曲线')            axes[0, 0].set_xlabel('交易次数')            axes[0, 0].set_ylabel('资金（元）')        # 2. 收益率分布        if '收益率' in trade_history.columns:            returns = trade_history[trade_history['操作'] == '卖出']['收益率']            axes[0, 1].hist(returns, bins=20, alpha=0.7)            axes[0, 1].axvline(x=returns.mean(), color='r', linestyle='--', label=f'平均: {returns.mean():.2%}')            axes[0, 1].set_title('单次交易收益率分布')            axes[0, 1].set_xlabel('收益率')            axes[0, 1].legend()        # 3. 月度收益热力图        # 这里需要更长时间的数据，暂时用模拟数据        months = ['1月', '2月', '3月', '4月', '5月', '6月']        years = ['2023', '2024']        monthly_returns = np.random.randn(len(years), len(months)) * 0.1        im = axes[1, 0].imshow(monthly_returns, cmap='RdYlGn', aspect='auto')        axes[1, 0].set_xticks(range(len(months)))        axes[1, 0].set_xticklabels(months)        axes[1, 0].set_yticks(range(len(years)))        axes[1, 0].set_yticklabels(years)        axes[1, 0].set_title('月度收益热力图')        plt.colorbar(im, ax=axes[1, 0])        # 4. 策略指标汇总        metrics = {            '总收益率': f"{backtest_results['总收益率']:.2%}",            '交易次数': backtest_results['交易次数'],            '胜率': '需计算',            '最大回撤': '需计算',            '夏普比率': '需计算'        }        axes[1, 1].axis('off')        table_data = [[k, v] for k, v in metrics.items()]        table = axes[1, 1].table(cellText=table_data,                                  colLabels=['指标', '数值'],                                 loc='center',                                 cellLoc='center')        table.auto_set_font_size(False)        table.set_fontsize(10)        table.scale(1, 2)        axes[1, 1].set_title('策略绩效指标')        plt.tight_layout()        plt.show()

def complete_analysis_pipeline():    """完整的分析流程"""    # 1. 初始化分析器    analyzer = XueqiuHotStockAnalyzer()    strategy = HotStockMomentumStrategy(initial_capital=100000)    visualizer = XueqiuVisualizer()    # 2. 获取数据    print("正在获取雪球热股榜单数据...")    hot_stocks = analyzer.get_hot_stock_list()    if hot_stocks is not None:        print(f"获取到 {len(hot_stocks)} 只热门股票")        # 3. 计算热度因子        print("计算热度因子...")        hot_stocks_with_factors = analyzer.calculate_hotness_factor(hot_stocks)        # 4. 行业分析        print("进行行业分布分析...")        sector_analysis = analyzer.analyze_sector_distribution(hot_stocks_with_factors)        # 5. 可视化展示        print("生成可视化图表...")        visualizer.plot_hot_stock_ranking(hot_stocks_with_factors)        # 6. 策略回测（需要实际价格数据）        print("进行策略回测...")        # 这里需要真实的股价数据，以下为示例        # price_data = get_real_price_data(hot_stocks_with_factors['股票代码'].tolist())        # signals = strategy.generate_signals(hot_stocks_with_factors, price_data)        # results = strategy.backtest_strategy(signals, price_data)        # visualizer.plot_strategy_performance(results)        # 7. 输出分析报告        print("\n=== 分析报告 ===")        print(f"1. 最热门行业: {sector_analysis.index if sector_analysis isnotNoneelse'N/A'}")        print(f"2. 热度最高股票: {hot_stocks_with_factors.iloc['股票名称']}")        print(f"3. 平均关注人数: {hot_stocks_with_factors['关注人数'].mean():.0f}")        print(f"4. 股票数量: {len(hot_stocks_with_factors)}")        return {            'hot_stocks': hot_stocks_with_factors,            'sector_analysis': sector_analysis,            # 'backtest_results': results        }    else:        print("数据获取失败")        return None# 运行完整分析if __name__ == "__main__":    results = complete_analysis_pipeline()

class MultiFactorStrategy:    def __init__(self):        self.factors = {}    def add_factor(self, name, factor_data, weight):        """添加因子"""        self.factors[name] = {            'data': factor_data,            'weight': weight        }    def calculate_composite_score(self):        """计算综合得分"""        composite_scores = None        for name, factor_info in self.factors.items():            factor_data = factor_info['data']            weight = factor_info['weight']            # 标准化因子数据            normalized = (factor_data - factor_data.mean()) / factor_data.std()            if composite_scores is None:                composite_scores = normalized * weight            else:                composite_scores += normalized * weight        return composite_scores

from sklearn.ensemble import RandomForestRegressorfrom sklearn.model_selection import train_test_splitclass HotnessPredictor:    def __init__(self):        self.model = RandomForestRegressor(n_estimators=100, random_state=42)    def prepare_features(self, stock_data):        """准备特征数据"""        features = pd.DataFrame()        # 技术指标特征        features['price_change'] = stock_data['涨跌幅']        features['volume_ratio'] = stock_data['成交量'] / stock_data['成交量'].rolling(20).mean()        # 热度相关特征        features['attention_growth'] = stock_data['关注人数'].pct_change()        features['rank_change'] = stock_data['排名'].diff()        # 时间特征        features['day_of_week'] = stock_data.index.dayofweek        features['month'] = stock_data.index.month        return features.dropna()    def train_predict(self, features, target):        """训练并预测"""        X_train, X_test, y_train, y_test = train_test_split(            features, target, test_size=0.2, random_state=42        )        self.model.fit(X_train, y_train)        predictions = self.model.predict(X_test)        return predictions, self.model.feature_importances_