+
+
+
股票批量分析
+
+
+
+
+
API配置
+
+
+
+
+
+
+
+
+
如不填写，将使用系统默认配置
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
分析结果
+
+
+
+
+
+
+
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/tests/test_stream.py b/tests/test_stream.py
new file mode 100644
index 0000000..e501081
--- /dev/null
+++ b/tests/test_stream.py
@@ -0,0 +1,137 @@
+import os
+import requests
+import json
+from logger import get_logger, get_stream_logger
+from dotenv import load_dotenv
+
+# 获取日志器
+logger = get_logger()
+stream_logger = get_stream_logger()
+
+def test_api_stream():
+ """
+ 测试API流式响应功能
+ """
+ # 加载环境变量
+ load_dotenv()
+
+ # 获取API配置
+ api_url = os.getenv('API_URL')
+ api_key = os.getenv('API_KEY')
+ api_model = os.getenv('API_MODEL', 'gpt-3.5-turbo')
+
+ logger.info(f"开始测试API流式响应，API URL: {api_url}, MODEL: {api_model}")
+
+ # 检查API配置
+ if not api_url:
+ logger.error("API URL未配置，无法进行测试")
+ return
+
+ if not api_key:
+ logger.error("API Key未配置，无法进行测试")
+ return
+
+ # 标准化API URL
+ if not (api_url.endswith('/chat/completions') or api_url.endswith('/v1/chat/completions')):
+ if api_url.endswith('/v1'):
+ api_url = f"{api_url}/chat/completions"
+ elif api_url.endswith('/'):
+ api_url = f"{api_url}v1/chat/completions"
+ else:
+ api_url = f"{api_url}/v1/chat/completions"
+
+ logger.debug(f"标准化后的API URL: {api_url}")
+
+ # 构建简单的测试提示
+ prompt = "这是一个API流式响应测试。请给出一个简短的股票分析样例。"
+
+ # 构建请求头和请求体
+ headers = {
+ "Authorization": f"Bearer {api_key}",
+ "Content-Type": "application/json"
+ }
+
+ payload = {
+ "model": api_model,
+ "messages": [{"role": "user", "content": prompt}],
+ "stream": True # 明确设置stream参数为True
+ }
+
+ logger.debug(f"请求载荷: {json.dumps(payload, indent=2)}")
+
+ try:
+ logger.info(f"发起流式API请求: {api_url}")
+
+ response = requests.post(
+ api_url,
+ headers=headers,
+ json=payload,
+ timeout=60,
+ stream=True
+ )
+
+ logger.info(f"API流式响应状态码: {response.status_code}")
+ logger.debug(f"响应头: {response.headers}")
+
+ if response.status_code == 200:
+ logger.info("成功获取API流式响应，开始处理")
+
+ buffer = ""
+ chunk_count = 0
+
+ for line in response.iter_lines():
+ if line:
+ line_str = line.decode('utf-8')
+ logger.info(f"原始流式行: {line_str}")
+
+ # 跳过保持连接的空行
+ if line_str.strip() == '':
+ logger.debug("跳过空行")
+ continue
+
+ # 数据行通常以"data: "开头
+ if line_str.startswith('data: '):
+ data_content = line_str[6:].strip() # 移除 "data: " 前缀并去除前后空格
+ logger.info(f"数据内容: {data_content}")
+
+ # 检查是否为流的结束
+ if data_content == '[DONE]':
+ logger.info("收到流结束标记 [DONE]")
+ break
+
+ try:
+ # 解析JSON数据
+ json_data = json.loads(data_content)
+ logger.debug(f"JSON结构: {json.dumps(json_data, indent=2)}")
+
+ if 'choices' in json_data:
+ delta = json_data['choices'][0].get('delta', {})
+ content = delta.get('content', '')
+
+ if content:
+ chunk_count += 1
+ buffer += content
+ logger.info(f"内容片段 #{chunk_count}: {content}")
+ except json.JSONDecodeError as e:
+ logger.error(f"JSON解析错误: {e}, 内容: {data_content}")
+ else:
+ logger.warning(f"收到非'data:'开头的行: {line_str}")
+
+ logger.info(f"流式处理完成，共收到 {chunk_count} 个内容片段")
+ logger.info(f"完整内容:\n{buffer}")
+
+ else:
+ try:
+ error_response = response.json()
+ error_text = json.dumps(error_response, indent=2)
+ except:
+ error_text = response.text[:500] if response.text else "无响应内容"
+
+ logger.error(f"API请求失败: 状态码 {response.status_code}, 响应: {error_text}")
+
+ except Exception as e:
+ logger.error(f"测试过程中发生异常: {str(e)}")
+ logger.exception(e)
+
+if __name__ == "__main__":
+ test_api_stream()
diff --git a/web_server.py b/web_server.py
index a62105f..2c77137 100644
--- a/web_server.py
+++ b/web_server.py
@@ -1,8 +1,15 @@
-from flask import Flask, render_template, request, jsonify
+from flask import Flask, render_template, request, jsonify, Response, stream_with_context
from stock_analyzer import StockAnalyzer
from us_stock_service import USStockService
import threading
import os
+import traceback
+import requests
+from logger import get_logger, get_stream_logger
+
+# 获取日志器
+logger = get_logger()
+stream_logger = get_stream_logger()
app = Flask(__name__)
analyzer = StockAnalyzer()
@@ -11,27 +18,86 @@ us_stock_service = USStockService()
@app.route('/')
def index():
announcement = os.getenv('ANNOUNCEMENT_TEXT') or None
- return render_template('index.html', announcement=announcement)
+ # 获取默认API配置信息
+ default_api_url = os.getenv('API_URL', '')
+ default_api_model = os.getenv('API_MODEL', 'gpt-3.5-turbo')
+ # 不传递API_KEY到前端，出于安全考虑
+ return render_template('index.html',
+ announcement=announcement,
+ default_api_url=default_api_url,
+ default_api_model=default_api_model)
@app.route('/analyze', methods=['POST'])
def analyze():
try:
+ logger.info("开始处理分析请求")
data = request.json
stock_codes = data.get('stock_codes', [])
market_type = data.get('market_type', 'A')
+ logger.debug(f"接收到分析请求: stock_codes={stock_codes}, market_type={market_type}")
+
+ # 获取自定义API配置
+ custom_api_url = data.get('api_url')
+ custom_api_key = data.get('api_key')
+ custom_api_model = data.get('api_model')
+ custom_api_timeout = data.get('api_timeout', 60)
+
+ logger.debug(f"自定义API配置: URL={custom_api_url}, 模型={custom_api_model}, API Key={'已提供' if custom_api_key else '未提供'}")
+
+ # 创建新的分析器实例，使用自定义配置
+ custom_analyzer = StockAnalyzer(
+ custom_api_url=custom_api_url,
+ custom_api_key=custom_api_key,
+ custom_api_model=custom_api_model,
+ custom_api_timeout= custom_api_timeout,
+ )
+
if not stock_codes:
+ logger.warning("未提供股票代码")
return jsonify({'error': '请输入代码'}), 400
+
+ # 使用流式响应
+ def generate():
+ if len(stock_codes) == 1:
+ # 单个股票分析流式处理
+ stock_code = stock_codes[0].strip()
+ logger.info(f"开始单股流式分析: {stock_code}")
+
+ stream_logger.info(f"初始化单股分析流: {stock_code}")
+ init_message = f'{{"stream_type": "single", "stock_code": "{stock_code}"}}\n'
+ stream_logger.info(f"发送初始化消息: {init_message}")
+ yield init_message
+
+ for chunk in custom_analyzer.analyze_stock(stock_code, market_type, stream=True):
+ stream_logger.info(f"流式输出块: {chunk}")
+ yield chunk + '\n'
+ else:
+ # 批量分析流式处理
+ logger.info(f"开始批量流式分析: {stock_codes}")
+
+ stream_logger.info(f"初始化批量分析流: {stock_codes}")
+ init_message = f'{{"stream_type": "batch", "stock_codes": {stock_codes}}}\n'
+ stream_logger.info(f"发送初始化消息: {init_message}")
+ yield init_message
+
+ for chunk in custom_analyzer.scan_market(
+ [code.strip() for code in stock_codes],
+ min_score=0,
+ market_type=market_type,
+ stream=True
+ ):
+ stream_logger.info(f"流式输出块: {chunk}")
+ yield chunk + '\n'
+
+ logger.info("成功创建流式响应生成器")
+ return Response(stream_with_context(generate()), mimetype='application/json')
- results = []
- for stock_code in stock_codes:
- result = analyzer.analyze_stock(stock_code.strip(), market_type)
- results.append(result)
-
- return jsonify({'results': results})
except Exception as e:
- print(f"分析股票时出错: {str(e)}")
- return jsonify({'error': str(e)}), 500
+ error_msg = f"分析股票时出错: {str(e)}"
+ logger.error(error_msg)
+ logger.exception(e)
+ return jsonify({'error': error_msg}), 500
@app.route('/search_us_stocks', methods=['GET'])
def search_us_stocks():
@@ -47,8 +113,72 @@ def search_us_stocks():
print(f"搜索美股代码时出错: {str(e)}")
return jsonify({'error': str(e)}), 500
+@app.route('/test_api_connection', methods=['POST'])
+def test_api_connection():
+ """测试API连接"""
+ try:
+ logger.info("开始测试API连接")
+ data = request.json
+ api_url = data.get('api_url')
+ api_key = data.get('api_key')
+ api_model = data.get('api_model')
+
+ logger.debug(f"测试API连接: URL={api_url}, 模型={api_model}, API Key={'已提供' if api_key else '未提供'}")
+
+ if not api_url:
+ logger.warning("未提供API URL")
+ return jsonify({'error': '请提供API URL'}), 400
+
+ if not api_key:
+ logger.warning("未提供API Key")
+ return jsonify({'error': '请提供API Key'}), 400
+
+ # 构建API URL
+ test_url = api_url
+ if not (api_url.endswith('/chat/completions') or api_url.endswith('/v1/chat/completions')):
+ if api_url.endswith('/v1'):
+ test_url = f"{api_url}/chat/completions"
+ elif api_url.endswith('/'):
+ test_url = f"{api_url}chat/completions"
+ else:
+ test_url = f"{api_url}/v1/chat/completions"
+
+ logger.debug(f"完整API测试URL: {test_url}")
+
+ # 发送测试请求
+ response = requests.post(
+ test_url,
+ headers={
+ "Authorization": f"Bearer {api_key}",
+ "Content-Type": "application/json"
+ },
+ json={
+ "model": api_model or "gpt-3.5-turbo",
+ "messages": [
+ {"role": "user", "content": "Hello, this is a test message. Please respond with 'API connection successful'."}
+ ],
+ "max_tokens": 20
+ },
+ timeout=10
+ )
+
+ # 检查响应
+ if response.status_code == 200:
+ logger.info(f"API连接测试成功: {response.status_code}")
+ return jsonify({'success': True, 'message': '连接成功'})
+ else:
+ error_message = response.json().get('error', {}).get('message', '未知错误')
+ logger.warning(f"API连接测试失败: {response.status_code} - {error_message}")
+ return jsonify({'success': False, 'message': f'连接失败: {error_message}', 'status_code': response.status_code}), 400
+
+ except requests.exceptions.RequestException as e:
+ logger.error(f"API连接请求错误: {str(e)}")
+ return jsonify({'success': False, 'message': f'请求错误: {str(e)}'}), 400
+ except Exception as e:
+ logger.error(f"测试API连接时出错: {str(e)}")
+ logger.exception(e)
+ return jsonify({'success': False, 'message': f'测试连接时出错: {str(e)}'}), 500
+
if __name__ == '__main__':
- app.run(host='0.0.0.0', port=8888, debug=True)
-
-
-
\ No newline at end of file
+ logger.info("股票分析系统启动")
+ app.run(host='0.0.0.0', port=8888, debug=True)
\ No newline at end of file