#!/usr/bin/env python3 """ 使用东财概念板块数据分析本周强势板块 需要5000积分 """ import sys from pathlib import Path import pandas as pd from datetime import datetime, timedelta # 添加项目根目录到路径 current_dir = Path(__file__).parent sys.path.insert(0, str(current_dir)) from src.data.tushare_fetcher import TushareFetcher from loguru import logger def get_recent_trading_dates(days_back=5): """获取最近的交易日期""" dates = [] current = datetime.now() while len(dates) < days_back: # 排除周末 if current.weekday() < 5: # 0-4是周一到周五 dates.append(current.strftime('%Y%m%d')) current -= timedelta(days=1) return sorted(dates) # 升序返回 def analyze_eastmoney_concepts(fetcher: TushareFetcher): """使用东财概念板块数据分析""" try: if not fetcher.pro: logger.error("需要Tushare Pro权限") return logger.info("🚀 使用东财概念板块数据分析...") # 获取最近5个交易日 trading_dates = get_recent_trading_dates(5) logger.info(f"分析时间范围: {trading_dates[0]} 到 {trading_dates[-1]}") # 获取最新交易日的概念板块数据 latest_date = trading_dates[-1] try: # 使用东财概念板块接口 dc_concepts = fetcher.pro.dc_index(trade_date=latest_date) logger.info(f"获取到 {len(dc_concepts)} 个东财概念板块") if dc_concepts.empty: logger.warning("未获取到东财概念板块数据") return # 打印数据结构以便调试 logger.info(f"数据列名: {list(dc_concepts.columns)}") if not dc_concepts.empty: logger.info(f"样本数据:\n{dc_concepts.head(2)}") # 检查涨跌幅字段名 change_col = None for col in ['pct_chg', 'pct_change', 'change_pct', 'chg_pct']: if col in dc_concepts.columns: change_col = col break if change_col: # 按涨跌幅排序 dc_concepts = dc_concepts.sort_values(change_col, ascending=False) else: logger.warning("未找到涨跌幅字段，使用原始顺序") change_col = 'code' # 使用code作为默认排序 print("\n" + "="*80) print("📈 东财概念板块实时排行榜") print("="*80) # 显示表头 if change_col != 'code': print(f"{'排名':<4} {'概念名称':<25} {'涨跌幅':<10} {'概念代码':<15}") else: print(f"{'排名':<4} {'概念名称':<25} {'概念代码':<15}") print("-" * 80) for i, (_, concept) in enumerate(dc_concepts.head(20).iterrows()): rank = i + 1 name = concept.get('name', 'N/A')[:23] + '..' if len(str(concept.get('name', 'N/A'))) > 23 else concept.get('name', 'N/A') code = concept.get('ts_code', 'N/A') if change_col != 'code': change_pct = f"{concept[change_col]:+.2f}%" if not pd.isna(concept.get(change_col, 0)) else "N/A" print(f"{rank:<4} {name:<25} {change_pct:<10} {code:<15}") else: print(f"{rank:<4} {name:<25} {code:<15}") # 强势概念TOP10 if change_col != 'code': print(f"\n🚀 强势概念板块TOP10:") for i, (_, concept) in enumerate(dc_concepts.head(10).iterrows()): change_val = concept.get(change_col, 0) if not pd.isna(change_val): print(f" {i+1:2d}. {concept.get('name', 'N/A')}: {change_val:+.2f}%") # 弱势概念TOP10 print(f"\n📉 弱势概念板块TOP10:") weak_concepts = dc_concepts.tail(10).iloc[::-1] # 反转顺序 for i, (_, concept) in enumerate(weak_concepts.iterrows()): change_val = concept.get(change_col, 0) if not pd.isna(change_val): print(f" {i+1:2d}. {concept.get('name', 'N/A')}: {change_val:+.2f}%") else: print(f"\n📋 概念板块列表（前10个）:") for i, (_, concept) in enumerate(dc_concepts.head(10).iterrows()): print(f" {i+1:2d}. {concept.get('name', 'N/A')} ({concept.get('ts_code', 'N/A')})") return dc_concepts except Exception as e: logger.error(f"获取东财概念板块数据失败: {e}") return None except Exception as e: logger.error(f"分析东财概念板块失败: {e}") return None def analyze_concept_trend(fetcher: TushareFetcher, concept_codes=None): """分析概念板块的趋势（多日对比）""" try: if not fetcher.pro: logger.error("需要Tushare Pro权限") return logger.info("📊 分析概念板块趋势...") # 获取最近5个交易日 trading_dates = get_recent_trading_dates(5) # 如果没有指定概念代码，获取当日表现最好的前10个 if concept_codes is None: latest_concepts = analyze_eastmoney_concepts(fetcher) if latest_concepts is not None and not latest_concepts.empty: concept_codes = latest_concepts.head(5)['code'].tolist() else: logger.warning("无法获取概念代码") return print(f"\n" + "="*80) print("📈 热门概念板块多日趋势分析") print("="*80) for concept_code in concept_codes: concept_trend = [] for date in trading_dates: try: # 获取特定日期的概念数据 daily_data = fetcher.pro.dc_index( trade_date=date, ts_code=concept_code ) if not daily_data.empty: # 检查数据结构 logger.debug(f"概念 {concept_code} 在 {date} 的数据字段: {list(daily_data.columns)}") # 东财概念数据可能没有close字段，使用其他字段替代 close_value = daily_data.iloc[0].get('total_mv', 1) # 使用总市值代替 if close_value == 0: close_value = 1 # 避免除零 concept_trend.append({ 'date': date, 'name': daily_data.iloc[0]['name'], 'close': close_value, 'pct_chg': daily_data.iloc[0]['pct_change'] }) except Exception as e: logger.debug(f"获取概念 {concept_code} 在 {date} 的数据失败: {e}") continue # 输出趋势 if concept_trend: concept_name = concept_trend[0]['name'] print(f"\n📊 {concept_name} ({concept_code}) 近5日走势:") # 计算总涨跌幅 if len(concept_trend) >= 2: start_close = concept_trend[0]['close'] end_close = concept_trend[-1]['close'] if start_close != 0 and start_close is not None: total_change = (end_close - start_close) / start_close * 100 print(f" 总涨跌幅: {total_change:+.2f}%") else: print(f" 总涨跌幅: 无法计算（起始值为0）") # 显示每日数据 for data in concept_trend: print(f" {data['date']}: {data['pct_chg']:+6.2f}% (指数: {data['close']:8.2f})") print("\n" + "="*80) except Exception as e: logger.error(f"分析概念趋势失败: {e}") def get_concept_constituents(fetcher: TushareFetcher, concept_code: str): """获取概念板块成分股""" try: if not fetcher.pro: logger.error("需要Tushare Pro权限") return logger.info(f"获取概念 {concept_code} 的成分股...") # 尝试通过概念板块获取成分股 try: # 使用concept_detail接口（如果可用） constituents = fetcher.pro.concept_detail(id=concept_code) if not constituents.empty: print(f"\n📋 概念成分股 ({len(constituents)}只):") for _, stock in constituents.head(10).iterrows(): print(f" {stock['ts_code']}: {stock.get('name', 'N/A')}") else: logger.warning(f"概念 {concept_code} 无成分股数据") except Exception as e: logger.error(f"获取概念成分股失败: {e}") except Exception as e: logger.error(f"获取概念成分股失败: {e}") def main(): """主函数""" logger.info("🚀 开始使用东财概念板块数据分析...") # 初始化Tushare数据获取器 token = "0ed6419a00d8923dc19c0b58fc92d94c9a0696949ab91a13aa58a0cc" fetcher = TushareFetcher(token=token) # 1. 分析当日概念板块表现 concepts_data = analyze_eastmoney_concepts(fetcher) # 2. 分析热门概念的多日趋势 if concepts_data is not None and not concepts_data.empty: print("\n" + "="*80 + "\n") # 获取表现最好的前3个概念进行趋势分析 top_concepts = concepts_data.head(3)['ts_code'].tolist() analyze_concept_trend(fetcher, top_concepts) # 3. 获取第一个概念的成分股示例 # top_concept_code = top_concepts[0] if top_concepts else None # if top_concept_code: # get_concept_constituents(fetcher, top_concept_code) logger.info("✅ 分析完成!") if __name__ == "__main__": main()