import akshare as ak import numpy as np import pandas as pd import matplotlib.pyplot as plt from datetime import datetime, timedelta import requests import json import time import os class StockDataLoader: """股票数据加载器""" def __init__(self): self.cache = {} self.cache_time = {} def get_stock_data(self, stock_code, refresh=False): """获取股票基础数据""" now = datetime.now() # 如果缓存中有数据且未过期（15分钟内），则使用缓存 if not refresh and stock_code in self.cache and \ (now - self.cache_time.get(stock_code, datetime.min)).total_seconds() < 900: return self.cache[stock_code] try: # 使用akshare获取股票数据 # 日K线数据 df_daily = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=(now - timedelta(days=365)).strftime("%Y%m%d"), end_date=now.strftime("%Y%m%d"), adjust="qfq") # 实时行情 df_realtime = ak.stock_zh_a_spot_em() df_realtime = df_realtime[df_realtime['代码'] == stock_code] # 资金流向 df_flow = ak.stock_individual_fund_flow(stock=stock_code) # 基本面数据 df_fundamental = ak.stock_financial_report_sina(stock=stock_code, symbol="现金流量表") # 整合数据 data = { 'daily': df_daily, 'realtime': df_realtime, 'flow': df_flow, 'fundamental': df_fundamental, 'timestamp': now } # 更新缓存 self.cache[stock_code] = data self.cache_time[stock_code] = now return data except Exception as e: print(f"获取股票数据出错: {e}") return None def get_market_data(self): """获取市场整体数据""" try: # 指数数据 df_index = ak.stock_zh_index_spot() # 行业板块 df_industry = ak.stock_sector_spot(indicator="最新") # 北向资金 df_north = ak.stock_em_hsgt_north_net_flow_in(indicator="沪股通") # 市场资金 df_market_fund = ak.stock_market_fund_flow() return { 'index': df_index, 'industry': df_industry, 'north': df_north, 'market_fund': df_market_fund, 'timestamp': datetime.now() } except Exception as e: print(f"获取市场数据出错: {e}") return None class TechnicalAnalyzer: """技术分析器""" @staticmethod def calculate_ma(data, periods=[5, 10, 20, 60]): """计算移动平均线""" df = data.copy() for period in periods: df[f'MA{period}'] = df['收盘'].rolling(window=period).mean() return df @staticmethod def calculate_macd(data, fast=12, slow=26, signal=9): """计算MACD指标""" df = data.copy() # 计算EMA df['EMA_fast'] = df['收盘'].ewm(span=fast, adjust=False).mean() df['EMA_slow'] = df['收盘'].ewm(span=slow, adjust=False).mean() # 计算DIF、DEA和MACD df['DIF'] = df['EMA_fast'] - df['EMA_slow'] df['DEA'] = df['DIF'].ewm(span=signal, adjust=False).mean() df['MACD'] = 2 * (df['DIF'] - df['DEA']) return df @staticmethod def calculate_rsi(data, periods=[6, 12, 24]): """计算RSI指标""" df = data.copy() for period in periods: delta = df['收盘'].diff() gain = delta.where(delta > 0, 0) loss = -delta.where(delta < 0, 0) avg_gain = gain.rolling(window=period).mean() avg_loss = loss.rolling(window=period).mean() rs = avg_gain / avg_loss df[f'RSI{period}'] = 100 - (100 / (1 + rs)) return df @staticmethod def calculate_kdj(data, n=9, m1=3, m2=3): """计算KDJ指标""" df = data.copy() low_min = df['最低'].rolling(window=n).min() high_max = df['最高'].rolling(window=n).max() df['RSV'] = 100 * ((df['收盘'] - low_min) / (high_max - low_min)) df['K'] = df['RSV'].ewm(alpha=1/m1, adjust=False).mean() df['D'] = df['K'].ewm(alpha=1/m2, adjust=False).mean() df['J'] = 3 * df['K'] - 2 * df['D'] return df @staticmethod def calculate_boll(data, n=20, k=2): """计算布林带""" df = data.copy() df['BOLL_MID'] = df['收盘'].rolling(window=n).mean() df['BOLL_STD'] = df['收盘'].rolling(window=n).std() df['BOLL_UP'] = df['BOLL_MID'] + k * df['BOLL_STD'] df['BOLL_DOWN'] = df['BOLL_MID'] - k * df['BOLL_STD'] return df @staticmethod def get_technical_signals(data): """获取技术信号""" df = data.copy() # 计算各种技术指标 df = TechnicalAnalyzer.calculate_ma(df) df = TechnicalAnalyzer.calculate_macd(df) df = TechnicalAnalyzer.calculate_rsi(df) df = TechnicalAnalyzer.calculate_kdj(df) df = TechnicalAnalyzer.calculate_boll(df) # 获取最新数据 latest = df.iloc[-1] # 生成信号 signals = { 'trend': 'up' if latest['收盘'] > latest['MA20'] else 'down', 'ma_cross': 'golden' if latest['MA5'] > latest['MA10'] and df.iloc[-2]['MA5'] <= df.iloc[-2]['MA10'] else 'death' if latest['MA5'] < latest['MA10'] and df.iloc[-2]['MA5'] >= df.iloc[-2]['MA10'] else 'none', 'macd_signal': 'golden' if latest['DIF'] > latest['DEA'] and df.iloc[-2]['DIF'] <= df.iloc[-2]['DEA'] else 'death' if latest['DIF'] < latest['DEA'] and df.iloc[-2]['DIF'] >= df.iloc[-2]['DEA'] else 'none', 'rsi_status': 'overbought' if latest['RSI6'] > 70 else 'oversold' if latest['RSI6'] < 30 else 'normal', 'kdj_signal': 'golden' if latest['K'] > latest['D'] and df.iloc[-2]['K'] <= df.iloc[-2]['D'] else 'death' if latest['K'] < latest['D'] and df.iloc[-2]['K'] >= df.iloc[-2]['D'] else 'none', 'boll_position': 'upper' if latest['收盘'] > latest['BOLL_UP'] else 'lower' if latest['收盘'] < latest['BOLL_DOWN'] else 'middle' } return signals, df class FundamentalAnalyzer: """基本面分析器""" @staticmethod def analyze_financial_data(data): """分析财务数据""" try: # 提取关键财务指标 df = data['fundamental'] # 计算增长率 revenue_growth = None profit_growth = None if len(df) >= 8: # 确保有足够的数据计算同比增长 # 假设第一列是报告期，按时间排序 df = df.sort_values(by=df.columns[0], ascending=False) # 获取最新和去年同期的营收和利润 latest_revenue = float(df.iloc[0]['营业收入'].replace(',', '')) prev_revenue = float(df.iloc[4]['营业收入'].replace(',', '')) latest_profit = float(df.iloc[0]['净利润'].replace(',', '')) prev_profit = float(df.iloc[4]['净利润'].replace(',', '')) # 计算增长率 revenue_growth = (latest_revenue - prev_revenue) / prev_revenue * 100 profit_growth = (latest_profit - prev_profit) / prev_profit * 100 # 获取估值数据 valuation_data = ak.stock_a_lg_indicator(symbol=data['realtime'].iloc[0]['代码']) # 提取PE、PB等指标 pe = valuation_data['pe'].iloc[0] if 'pe' in valuation_data.columns else None pb = valuation_data['pb'].iloc[0] if 'pb' in valuation_data.columns else None ps = valuation_data['ps'].iloc[0] if 'ps' in valuation_data.columns else None # 获取行业平均估值 industry_code = valuation_data['industry'].iloc[0] if 'industry' in valuation_data.columns else None industry_valuation = None if industry_code: industry_stocks = ak.stock_sector_detail(sector=industry_code) industry_codes = industry_stocks['代码'].tolist() # 获取行业内股票的估值数据 industry_pe_list = [] industry_pb_list = [] for code in industry_codes[:10]: # 取前10只股票计算平均值 try: ind_valuation = ak.stock_a_lg_indicator(symbol=code) if 'pe' in ind_valuation.columns and not pd.isna(ind_valuation['pe'].iloc[0]): industry_pe_list.append(ind_valuation['pe'].iloc[0]) if 'pb' in ind_valuation.columns and not pd.isna(ind_valuation['pb'].iloc[0]): industry_pb_list.append(ind_valuation['pb'].iloc[0]) except: continue industry_pe = sum(industry_pe_list) / len(industry_pe_list) if industry_pe_list else None industry_pb = sum(industry_pb_list) / len(industry_pb_list) if industry_pb_list else None industry_valuation = { 'pe': industry_pe, 'pb': industry_pb } return { 'growth': { 'revenue': revenue_growth, 'profit': profit_growth }, 'valuation': { 'pe': pe, 'pb': pb, 'ps': ps }, 'industry_valuation': industry_valuation } except Exception as e: print(f"分析基本面数据出错: {e}") return None class CapitalAnalyzer: """资金面分析器""" @staticmethod def analyze_capital_flow(data): """分析资金流向""" try: # 主力资金流向 df_flow = data['flow'] # 计算近5日主力净流入 recent_days = min(5, len(df_flow)) recent_flow = df_flow.iloc[:recent_days] main_net_inflow = recent_flow['主力净流入'].sum() # 计算主力净流入占比 if '成交额' in recent_flow.columns: total_amount = recent_flow['成交额'].sum() main_ratio = main_net_inflow / total_amount if total_amount else 0 else: main_ratio = None # 获取北向资金持股 try: north_holding = ak.stock_em_hsgt_hold_stock() stock_north = north_holding[north_holding['代码'] == data['realtime'].iloc[0]['代码']] if not stock_north.empty: north_amount = stock_north['持股市值'].iloc[0] north_ratio = stock_north['持股占比'].iloc[0] else: north_amount = None north_ratio = None except: north_amount = None north_ratio = None # 获取机构持股 try: inst_holding = ak.stock_institute_hold(stock=data['realtime'].iloc[0]['代码']) inst_ratio = inst_holding['持股占流通股比例'].iloc[0] if not inst_holding.empty else None except: inst_ratio = None return { 'main_flow': { 'net_inflow': main_net_inflow, 'ratio': main_ratio }, 'north_holding': { 'amount': north_amount, 'ratio': north_ratio }, 'inst_holding': inst_ratio } except Exception as e: print(f"分析资金流向出错: {e}") return None class PolicyAnalyzer: """政策分析器""" def __init__(self, api_key=None): self.api_key = api_key self.deepseek_url = "https://api.deepseek.com/v1/chat/completions" def get_policy_news(self): """获取政策新闻""" try: # 获取财经新闻 news = ak.stock_news_em() # 筛选政策相关新闻 policy_keywords = ['政策', '监管', '证监会', '央行', '银保监会', '发改委', '财政部'] policy_news = news[news['新闻标题'].str.contains('|'.join(policy_keywords))] return policy_news.iloc[:10] if len(policy_news) > 0 else news.iloc[:5] except Exception as e: print(f"获取政策新闻出错: {e}") return None def analyze_policy_impact(self, stock_code, industry): """分析政策影响""" news = self.get_policy_news() if news is None or len(news) == 0: return { 'impact_score': 50, # 中性 'summary': "无法获取政策新闻" } # 提取新闻标题和内容 news_texts = [] for _, row in news.iterrows(): news_texts.append(f"标题: {row['新闻标题']}\n内容: {row['新闻内容'] if '新闻内容' in row else '无详细内容'}") news_text = "\n\n".join(news_texts) # 如果有API密钥，使用DeepSeek API分析 if self.api_key: prompt = f""" 请分析以下最新财经政策新闻对股票 {stock_code}（行业：{industry}）的影响: {news_text} 请输出: 1. 影响评分(0-100，0表示极度负面，50表示中性，100表示极度正面) 2. 影响总结(100字以内) 格式: {{ "impact_score": 评分, "summary": "总结内容" }} """ try: headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } payload = { "model": "deepseek-chat", "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 500 } response = requests.post(self.deepseek_url, headers=headers, json=payload) if response.status_code == 200: result = response.json() content = result['choices'][0]['message']['content'] try: # 尝试解析JSON响应 analysis = json.loads(content) return analysis except: # 如果解析失败，返回默认值 return { 'impact_score': 50, 'summary': "API返回格式错误，无法解析" } else: print(f"API调用失败: {response.status_code} - {response.text}") except Exception as e: print(f"调用DeepSeek API出错: {e}") # 如果没有API密钥或API调用失败，返回简单分析 return { 'impact_score': 50, # 中性 'summary': "无法进行深度政策分析，请配置DeepSeek API密钥" } class AIAdvisor: """AI决策顾问""" def __init__(self, api_key=None): self.api_key = api_key self.deepseek_url = "https://api.deepseek.com/v1/chat/completions" def generate_decision(self, stock_data, technical_analysis, fundamental_analysis, capital_analysis, policy_analysis): """生成AI决策""" # 准备分析数据 stock_code = stock_data['realtime'].iloc[0]['代码'] stock_name = stock_data['realtime'].iloc[0]['名称'] current_price = stock_data['realtime'].iloc[0]['最新价'] change_pct = stock_data['realtime'].iloc[0]['涨跌幅'] # 技术面数据 tech_signals = technical_analysis[0] # 技术信号 # 基本面数据 growth = fundamental_analysis['growth'] if fundamental_analysis else {'revenue': None, 'profit': None} valuation = fundamental_analysis['valuation'] if fundamental_analysis else {'pe': None, 'pb': None} # 资金面数据 capital_flow = capital_analysis['main_flow'] if capital_analysis else {'net_inflow': None} # 政策面数据 policy_impact = policy_analysis['impact_score'] if policy_analysis else 50 # 如果有API密钥，使用DeepSeek API生成决策 if self.api_key: prompt = f""" 作为专业股票分析师，请基于以下数据为股票 {stock_code}（{stock_name}）生成交易决策: [基本信息] 当前价格: {current_price} 涨跌幅: {change_pct}% [技术面] 趋势: {tech_signals['trend']} 均线交叉: {tech_signals['ma_cross']} MACD信号: {tech_signals['macd_signal']} RSI状态: {tech_signals['rsi_status']} KDJ信号: {tech_signals['kdj_signal']} 布林带位置: {tech_signals['boll_position']} [基本面] 营收增长: {growth['revenue']}% 利润增长: {growth['profit']}% 市盈率(PE): {valuation['pe']} 市净率(PB): {valuation['pb']} [资金面] 主力净流入: {capital_flow['net_inflow']}万元 [政策面] 政策影响评分(0-100): {policy_impact} 请按以下格式输出决策: {{ "action": "买入/持有/卖出", "target_price": 目标价, "stop_loss": 止损价, "confidence": 置信度(0-100), "holding_days": 建议持仓天数, "reasons": ["理由1", "理由2", "理由3"], "risks": ["风险1", "风险2"] }} 注意: 1. 目标年化收益40-60% 2. 小资金单票策略 3. 可以右侧交易，部分确定的左侧 """ try: headers = { "Content-Type": "application/json", "Authorization": f"Bearer {self.api_key}" } payload = { "model": "deepseek-chat", "messages": [ {"role": "user", "content": prompt} ], "temperature": 0.3, "max_tokens": 800 } response = requests.post(self.deepseek_url, headers=headers, json=payload) if response.status_code == 200: result = response.json() content = result['choices'][0]['message']['content'] try: # 尝试解析JSON响应 decision = json.loads(content) return decision except: # 如果解析失败，返回默认决策 print(f"无法解析API响应: {content}") else: print(f"API调用失败: {response.status_code} - {response.text}") except Exception as e: print(f"调用DeepSeek API出错: {e}") # 如果没有API密钥或API调用失败，使用规则生成简单决策 return self._rule_based_decision(stock_data, technical_analysis, fundamental_analysis, capital_analysis, policy_analysis) def _rule_based_decision(self, stock_data, technical_analysis, fundamental_analysis, capital_analysis, policy_analysis): """基于规则的决策生成""" # 技术面信号 tech_signals = technical_analysis[0] # 基本信息 current_price = stock_data['realtime'].iloc[0]['最新价'] # 计算技术面得分 tech_score = 0 if tech_signals['trend'] == 'up': tech_score += 20 if tech_signals['ma_cross'] == 'golden': tech_score += 15 elif tech_signals['ma_cross'] == 'death': tech_score -= 15 if tech_signals['macd_signal'] == 'golden': tech_score += 15 elif tech_signals['macd_signal'] == 'death': tech_score -= 15 if tech_signals['rsi_status'] == 'oversold': tech_score += 10 elif tech_signals['rsi_status'] == 'overbought': tech_score -= 10 if tech_signals['kdj_signal'] == 'golden': tech_score += 10 elif tech_signals['kdj_signal'] == 'death': tech_score -= 10 # 基本面得分 fund_score = 0 if fundamental_analysis: if fundamental_analysis['growth']['revenue'] and fundamental_analysis['growth']['revenue'] > 20: fund_score += 15 if fundamental_analysis['growth']['profit'] and fundamental_analysis['growth']['profit'] > 20: fund_score += 15 # 估值比较 if fundamental_analysis['valuation']['pe'] and fundamental_analysis['industry_valuation'] and fundamental_analysis['industry_valuation']['pe']: if fundamental_analysis['valuation']['pe'] < fundamental_analysis['industry_valuation']['pe'] * 0.8: fund_score += 10 elif fundamental_analysis['valuation']['pe'] > fundamental_analysis['industry_valuation']['pe'] * 1.2: fund_score -= 10 # 资金面得分 capital_score = 0 if capital_analysis and capital_analysis['main_flow']['net_inflow']: if capital_analysis['main_flow']['net_inflow'] > 1000: # 1000万 capital_score += 20 elif capital_analysis['main_flow']['net_inflow'] < -1000: capital_score -= 20 # 政策面得分 policy_score = 0 if policy_analysis: policy_score = (policy_analysis['impact_score'] - 50) / 2.5 # 转换为-20到20的范围 # 总分 total_score = tech_score + fund_score + capital_score + policy_score # 决策 if total_score > 30: action = "买入" confidence = min(60 + total_score / 2, 95) target_price = current_price * 1.15 stop_loss = current_price * 0.92 holding_days = 14 elif total_score < -30: action = "卖出" confidence = min(60 + abs(total_score) / 2, 95) target_price = current_price * 0.9 stop_loss = current_price * 1.05 holding_days = 7 else: action = "持有" confidence = 50 + abs(total_score) / 2 target_price = current_price * (1.05 if total_score > 0 else 0.98) stop_loss = current_price * (0.95 if total_score > 0 else 0.9) holding_days = 10 # 生成理由 reasons = [] risks = [] if tech_signals['trend'] == 'up': reasons.append("股价处于上升趋势") else: risks.append("股价处于下降趋势") if tech_signals['ma_cross'] == 'golden': reasons.append("均线金叉，短期看涨") elif tech_signals['ma_cross'] == 'death': risks.append("均线死叉，短期看跌") if tech_signals['macd_signal'] == 'golden': reasons.append("MACD金叉，买入信号") elif tech_signals['macd_signal'] == 'death': risks.append("MACD死叉，卖出信号") if fundamental_analysis and fundamental_analysis['growth']['profit'] and fundamental_analysis['growth']['profit'] > 20: reasons.append(f"利润增长强劲，同比增长{fundamental_analysis['growth']['profit']:.1f}%") if capital_analysis and capital_analysis['main_flow']['net_inflow'] and capital_analysis['main_flow']['net_inflow'] > 1000: reasons.append(f"主力资金净流入{capital_analysis['main_flow']['net_inflow']:.0f}万元") elif capital_analysis and capital_analysis['main_flow']['net_inflow'] and capital_analysis['main_flow']['net_inflow'] < -1000: risks.append(f"主力资金净流出{abs(capital_analysis['main_flow']['net_inflow']):.0f}万元") if policy_analysis and policy_analysis['impact_score'] > 60: reasons.append(f"政策面利好: {policy_analysis['summary']}") elif policy_analysis and policy_analysis['impact_score'] < 40: risks.append(f"政策面利空: {policy_analysis['summary']}") # 确保至少有一个理由和风险 if not reasons: reasons.append("技术面和基本面综合评分适中") if not risks: risks.append("市场波动风险") risks.append("大盘系统性风险") return { "action": action, "target_price": round(target_price, 2), "stop_loss": round(stop_loss, 2), "confidence": round(confidence, 1), "holding_days": holding_days, "reasons": reasons[:3], # 最多3个理由 "risks": risks[:2] # 最多2个风险 } class StockScreener: """股票筛选器""" def __init__(self): self.data_loader = StockDataLoader() self.technical_analyzer = TechnicalAnalyzer() self.fundamental_analyzer = FundamentalAnalyzer() self.capital_analyzer = CapitalAnalyzer() def initial_screen(self, count=300): """初始筛选，获取符合基本条件的股票""" try: # 获取A股所有股票 all_stocks = ak.stock_zh_a_spot_em() # 基本筛选条件 # 1. 剔除ST股票 filtered = all_stocks[~all_stocks['名称'].str.contains('ST')] # 2. 剔除当日涨停或跌停的股票 filtered = filtered[ (filtered['涨跌幅'] < 9.5) & (filtered['涨跌幅'] > -9.5) ] # 3. 按成交量排序，剔除成交量过小的股票 filtered = filtered.sort_values(by='成交量', ascending=False) filtered = filtered.head(count) return filtered except Exception as e: print(f"初始筛选出错: {e}") return pd.DataFrame() def technical_filter(self, stocks, count=100): """技术面筛选""" result = [] for _, stock in stocks.iterrows(): try: # 获取股票数据 stock_data = self.data_loader.get_stock_data(stock['代码']) if stock_data is None: continue # 技术分析 tech_signals, tech_data = self.technical_analyzer.get_technical_signals(stock_data['daily']) # 技术面筛选条件 # 1. 股价在20日均线以上 if tech_signals['trend'] != 'up': continue # 2. MACD金叉或即将金叉 if tech_signals['macd_signal'] != 'golden' and tech_data.iloc[-1]['DIF'] < tech_data.iloc[-1]['DEA'] - 0.1: continue # 3. RSI不处于超买区 if tech_signals['rsi_status'] == 'overbought': continue # 计算技术面得分 tech_score = 0 if tech_signals['trend'] == 'up': tech_score += 20 if tech_signals['ma_cross'] == 'golden': tech_score += 15 if tech_signals['macd_signal'] == 'golden': tech_score += 15 if tech_signals['rsi_status'] == 'oversold': tech_score += 10 if tech_signals['kdj_signal'] == 'golden': tech_score += 10 result.append({ 'code': stock['代码'], 'name': stock['名称'], 'price': stock['最新价'], 'change': stock['涨跌幅'], 'tech_score': tech_score, 'tech_signals': tech_signals }) except Exception as e: print(f"技术面筛选股票 {stock['代码']} 出错: {e}") continue # 按技术面得分排序 result.sort(key=lambda x: x['tech_score'], reverse=True) return result[:count] def fundamental_filter(self, stocks, count=30): """基本面筛选""" result = [] for stock in stocks: try: # 获取股票数据 stock_data = self.data_loader.get_stock_data(stock['code']) if stock_data is None: continue # 基本面分析 fund_analysis = self.fundamental_analyzer.analyze_financial_data(stock_data) if fund_analysis is None: continue # 基本面筛选条件 # 1. 利润增长为正 if fund_analysis['growth']['profit'] is None or fund_analysis['growth']['profit'] <= 0: continue # 2. PE合理 if fund_analysis['valuation']['pe'] is None or fund_analysis['valuation']['pe'] > 50: continue # 计算基本面得分 fund_score = 0 if fund_analysis['growth']['revenue'] and fund_analysis['growth']['revenue'] > 20: fund_score += 15 if fund_analysis['growth']['profit'] and fund_analysis['growth']['profit'] > 20: fund_score += 15 # 估值比较 if fund_analysis['valuation']['pe'] and fund_analysis['industry_valuation'] and fund_analysis['industry_valuation']['pe']: if fund_analysis['valuation']['pe'] < fund_analysis['industry_valuation']['pe'] * 0.8: fund_score += 10 elif fund_analysis['valuation']['pe'] > fund_analysis['industry_valuation']['pe'] * 1.2: fund_score -= 10 result.append({ 'code': stock['code'], 'name': stock['name'], 'price': stock['price'], 'change': stock['change'], 'tech_score': stock['tech_score'], 'fund_score': fund_score, 'fund_analysis': fund_analysis }) except Exception as e: print(f"基本面筛选股票 {stock['code']} 出错: {e}") continue # 按基本面得分排序 result.sort(key=lambda x: x['fund_score'], reverse=True) return result[:count] def capital_filter(self, stocks, count=10): """资金面筛选""" result = [] for stock in stocks: try: # 获取股票数据 stock_data = self.data_loader.get_stock_data(stock['code']) if stock_data is None: continue # 资金面分析 capital_analysis = self.capital_analyzer.analyze_capital_flow(stock_data) if capital_analysis is None: continue # 计算资金面得分 capital_score = 0 if capital_analysis['main_flow']['net_inflow']: if capital_analysis['main_flow']['net_inflow'] > 1000: # 1000万 capital_score += 20 elif capital_analysis['main_flow']['net_inflow'] > 500: # 500万 capital_score += 10 elif capital_analysis['main_flow']['net_inflow'] < -1000: capital_score -= 20 elif capital_analysis['main_flow']['net_inflow'] < -500: capital_score -= 10 if capital_analysis['north_holding']['ratio']: if capital_analysis['north_holding']['ratio'] > 5: # 北向持股比例>5% capital_score += 10 if capital_analysis['inst_holding']: if capital_analysis['inst_holding'] > 50: # 机构持股比例>50% capital_score += 10 # 总分 = 技术面 + 基本面 + 资金面 total_score = stock['tech_score'] + stock['fund_score'] + capital_score result.append({ 'code': stock['code'], 'name': stock['name'], 'price': stock['price'], 'change': stock['change'], 'tech_score': stock['tech_score'], 'fund_score': stock['fund_score'], 'capital_score': capital_score, 'total_score': total_score, 'capital_analysis': capital_analysis }) except Exception as e: print(f"资金面筛选股票 {stock['code']} 出错: {e}") continue # 按总分排序 result.sort(key=lambda x: x['total_score'], reverse=True) return result[:count] def run_screening_pipeline(self): """运行完整筛选流程""" print("开始股票筛选流程...") # 1. 初始筛选 print("第一阶段: 初始筛选...") initial_stocks = self.initial_screen(300) print(f"初始筛选完成，获取 {len(initial_stocks)} 只股票") # 2. 技术面筛选 print("第二阶段: 技术面筛选...") tech_filtered = self.technical_filter(initial_stocks, 100) print(f"技术面筛选完成，获取 {len(tech_filtered)} 只股票") # 3. 基本面筛选 print("第三阶段: 基本面筛选...") fund_filtered = self.fundamental_filter(tech_filtered, 30) print(f"基本面筛选完成，获取 {len(fund_filtered)} 只股票") # 4. 资金面筛选 print("第四阶段: 资金面筛选...") final_stocks = self.capital_filter(fund_filtered, 10) print(f"资金面筛选完成，获取 {len(final_stocks)} 只股票") return final_stocks class StockMonitor: """股票监控器""" def __init__(self, api_key=None): self.data_loader = StockDataLoader() self.technical_analyzer = TechnicalAnalyzer() self.fundamental_analyzer = FundamentalAnalyzer() self.capital_analyzer = CapitalAnalyzer() self.policy_analyzer = PolicyAnalyzer(api_key) self.ai_advisor = AIAdvisor(api_key) self.last_check_time = {} def check_position(self, stock_code): """检查持仓股票状态""" # 获取股票数据 stock_data = self.data_loader.get_stock_data(stock_code, refresh=True) if stock_data is None: return { 'status': 'error', 'message': '获取股票数据失败' } # 技术分析 technical_analysis = self.technical_analyzer.get_technical_signals(stock_data['daily']) # 基本面分析 fundamental_analysis = self.fundamental_analyzer.analyze_financial_data(stock_data) # 资金面分析 capital_analysis = self.capital_analyzer.analyze_capital_flow(stock_data) # 政策面分析 industry = stock_data['realtime'].iloc[0]['所属行业'] if '所属行业' in stock_data['realtime'].columns else None policy_analysis = self.policy_analyzer.analyze_policy_impact(stock_code, industry) # 生成决策 decision = self.ai_advisor.generate_decision( stock_data, technical_analysis, fundamental_analysis, capital_analysis, policy_analysis ) # 更新检查时间 self.last_check_time[stock_code] = datetime.now() return { 'status': 'success', 'time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'stock_code': stock_code, 'stock_name': stock_data['realtime'].iloc[0]['名称'], 'price': stock_data['realtime'].iloc[0]['最新价'], 'change': stock_data['realtime'].iloc[0]['涨跌幅'], 'decision': decision } def start_monitoring(self, stock_codes, interval=900): """开始监控股票，每隔interval秒检查一次""" while True: current_time = datetime.now().time() # 判断是否在交易时间 if self._is_trading_hours(current_time): for stock_code in stock_codes: try: # 检查上次检查时间，避免频繁检查 last_time = self.last_check_time.get(stock_code) if last_time and (datetime.now() - last_time).total_seconds() < interval: continue result = self.check_position(stock_code) if result['status'] == 'success': print(f"[{result['time']}] 检查股票: {result['stock_name']}({result['stock_code']})") print(f"当前价格: {result['price']} ({result['change']}%)") print(f"AI决策: {result['decision']['action']}") print(f"目标价: {result['decision']['target_price']}, 止损价: {result['decision']['stop_loss']}") print(f"置信度: {result['decision']['confidence']}%") print(f"理由: {', '.join(result['decision']['reasons'])}") print(f"风险: {', '.join(result['decision']['risks'])}") print("-" * 50) else: print(f"检查股票 {stock_code} 失败: {result['message']}") except Exception as e: print(f"监控股票 {stock_code} 出错: {e}") # 休眠一段时间 time.sleep(60) # 每分钟检查一次交易时间 def _is_trading_hours(self, t): """判断是否在交易时间""" # 上午9:30-11:30，下午13:00-15:00 morning_start = datetime.strptime("9:30", "%H:%M").time() morning_end = datetime.strptime("11:30", "%H:%M").time() afternoon_start = datetime.strptime("13:00", "%H:%M").time() afternoon_end = datetime.strptime("15:00", "%H:%M").time() return (morning_start <= t <= morning_end) or (afternoon_start <= t <= afternoon_end) # 主程序 if __name__ == "__main__": # 设置DeepSeek API密钥 api_key = os.environ.get("DEEPSEEK_API_KEY") print("股票AI分析系统启动...") print("=" * 50) # 创建股票筛选器 screener = StockScreener() # 运行早盘筛选 top_stocks = screener.run_screening_pipeline() print("\n今日AI精选股票:") for i, stock in enumerate(top_stocks): print(f"{i+1}. {stock['name']}({stock['code']}) - 总分: {stock['total_score']}") print("\n选择一只股票进行详细分析...") selected_stock = top_stocks[0]['code'] # 默认选择第一只 # 创建股票监控器 monitor = StockMonitor(api_key) # 分析选中的股票 result = monitor.check_position(selected_stock) if result['status'] == 'success': print(f"\n股票详细分析: {result['stock_name']}({result['stock_code']})") print(f"当前价格: {result['price']} ({result['change']}%)") print(f"AI决策: {result['decision']['action']}") print(f"目标价: {result['decision']['target_price']}, 止损价: {result['decision']['stop_loss']}") print(f"置信度: {result['decision']['confidence']}%") print(f"建议持仓天数: {result['decision']['holding_days']}天") print("\n决策理由:") for reason in result['decision']['reasons']: print(f"- {reason}") print("\n风险提示:") for risk in result['decision']['risks']: print(f"- {risk}") print("\n开始实时监控...") # 监控选中的股票 monitor.start_monitoring([selected_stock])