TradingAgents AI 交易分析

"""TradingAgents Web 可视化页面提供Web界面让用户输入股票代码，运行TradingAgents分析，查看结果。 """ import json import os import sys import threading import uuid from datetime import datetime from flask import Flask, render_template_string, request, jsonify # 确保项目根目录在路径中 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) app = Flask(__name__) # 存储运行中的任务 tasks = {} # HTML模板 HTML_TEMPLATE = """ TradingAgents AI 交易分析

正在初始化分析...

📊 交易决策

📈 技术分析师

等待分析...

💭 情绪分析师

等待分析...

📰 新闻分析师

等待分析...

🔬 研究员辩论

等待分析...

⚠️ 风险评估

等待分析...

""" @app.route('/') def index(): default_date = datetime.now().strftime('%Y-%m-%d') return render_template_string(HTML_TEMPLATE, default_date=default_date) @app.route('/api/analyze', methods=['POST']) def analyze(): data = request.json ticker = data.get('ticker', '').strip() date = data.get('date', datetime.now().strftime('%Y-%m-%d')) if not ticker: return jsonify({'error': '请输入股票代码'}), 400 task_id = str(uuid.uuid4())[:8] tasks[task_id] = { 'status': 'running', 'ticker': ticker, 'date': date, 'logs': ['⏳ 初始化分析...'], 'result': None, 'error': None, 'started_at': datetime.now().isoformat() } # 在后台线程运行分析 thread = threading.Thread(target=run_analysis, args=(task_id, ticker, date)) thread.daemon = True thread.start() return jsonify({'task_id': task_id, 'status': 'running'}) @app.route('/api/status/') def task_status(task_id): task = tasks.get(task_id) if not task: return jsonify({'error': '任务不存在'}), 404 return jsonify(task) def run_analysis(task_id, ticker, date): """后台运行TradingAgents分析。""" import sys import io try: task = tasks[task_id] def log(msg): task['logs'].append(msg) # 保持日志不超过50条 if len(task['logs']) > 50: task['logs'] = task['logs'][-50:] # 设置环境变量 os.environ['OPENAI_COMPATIBLE_API_KEY'] = 'tp-ce6eaqg1gh1uev46i4oexh4sahloyfo2hfpxuc2n9m7yl9dd' os.environ['FRED_API_KEY'] = '825283926dd016ea66e96cc919e047da' os.environ['ALPHA_VANTAGE_API_KEY'] = 'W5PTA2J5OB16W02L' log(f'📊 开始分析 {ticker} ({date})') log(f'🤖 模型: MiMo-v2.5-pro') log(f'🌐 数据源: Alpha Vantage + SSH中转代理') from tradingagents.graph.trading_graph import TradingAgentsGraph from tradingagents.default_config import DEFAULT_CONFIG config = DEFAULT_CONFIG.copy() config['llm_provider'] = 'openai_compatible' config['backend_url'] = 'https://token-plan-cn.xiaomimimo.com/v1' config['deep_think_llm'] = 'mimo-v2.5-pro' config['quick_think_llm'] = 'mimo-v2.5-pro' config['max_debate_rounds'] = 1 config['max_risk_discuss_rounds'] = 1 # 配置数据源：Yahoo Finance优先，Alpha Vantage作为备选 config['data_vendors'] = { 'core_stock_apis': 'yfinance,alpha_vantage', 'technical_indicators': 'yfinance,alpha_vantage', 'fundamental_data': 'yfinance,alpha_vantage', 'news_data': 'yfinance,alpha_vantage', 'macro_data': 'fred', 'prediction_markets': 'polymarket', } log('🔧 配置完成，启动分析引擎...') log('⏳ 分析通常需要3-8分钟，请耐心等待...') # 捕获stdout输出以获取实时进度 class LogCapture: def __init__(self, task): self.task = task self.buffer = '' def write(self, text): self.buffer += text if '\n' in self.buffer: lines = self.buffer.split('\n') self.buffer = lines[-1] for line in lines[:-1]: line = line.strip() if line and len(line) > 5: # 只保留有意义的输出 if any(k in line for k in ['Tool', 'Message', 'Agent', 'failed', 'error', '✅', '❌', '📊', '📈', '💭', '📰', '🔬', '⚠️']): self.task['logs'].append(f'📋 {line[:100]}') if len(self.task['logs']) > 50: self.task['logs'] = self.task['logs'][-50:] def flush(self): pass # 临时捕获输出 old_stdout = sys.stdout sys.stdout = LogCapture(task) try: ta = TradingAgentsGraph(debug=True, config=config) log('📈 技术分析师开始工作...') state, decision = ta.propagate(ticker, date) finally: sys.stdout = old_stdout log(f'✅ 分析完成: {decision}') # 提取各Agent的报告 reports = {} # 从state中提取各阶段报告 if 'market_report' in state: reports['technical'] = state['market_report'][:2000] if 'sentiment_report' in state: reports['sentiment'] = state['sentiment_report'][:2000] if 'news_report' in state: reports['news'] = state['news_report'][:2000] if 'investment_debate_state' in state: debate = state['investment_debate_state'] if isinstance(debate, dict): reports['debate'] = str(debate.get('judge_decision', ''))[:2000] else: reports['debate'] = str(debate)[:2000] if 'risk_debate_state' in state: risk = state['risk_debate_state'] if isinstance(risk, dict): reports['risk'] = str(risk.get('judge_decision', ''))[:2000] else: reports['risk'] = str(risk)[:2000] task['status'] = 'completed' task['result'] = { 'final_decision': decision, 'reports': reports } task['logs'].append(f'✅ 最终决策: {decision}') except Exception as e: task = tasks.get(task_id) if task: task['status'] = 'failed' task['error'] = str(e) task['logs'].append(f'❌ 错误: {str(e)}') if __name__ == '__main__': app.run(host='0.0.0.0', port=8890, debug=False)

🤖 TradingAgents

📈 技术分析师

💭 情绪分析师

📰 新闻分析师

🔬 研究员辩论

⚠️ 风险评估