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分析。""" 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' log(f'📊 开始分析 {ticker} ({date})') log(f'🤖 模型: MiMo-v2.5-pro') log(f'🌐 数据源: 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 log('🔧 配置完成，启动分析引擎...') ta = TradingAgentsGraph(debug=True, config=config) log('📈 技术分析中...') state, decision = ta.propagate(ticker, date) 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

📈 技术分析师

💭 情绪分析师

📰 新闻分析师

🔬 研究员辩论

⚠️ 风险评估