perf: 6项性能与正确性优化
1. get_stock_name: 延迟批量保存缓存(60秒去抖),避免每次新缓存都写文件 2. get_stock_name: 优先从DB stock_realtime_price查name,未命中再调腾讯API 3. compute_comprehensive_score: 单股模式也使用_fund_flow_cache(30分钟TTL) 4. _get_kline_from_local_db: 移除conn.autocommit=True,避免污染连接池事务模式 5. deep_analyze: K线天数从180改为120,与实时检测统一 6. detect_all_signals: 已有指标列时跳过重复calc_all_indicators
This commit is contained in:
@@ -99,7 +99,7 @@ def deep_analyze():
|
||||
if not stock_code:
|
||||
return jsonify({'error': '股票代码不能为空'}), 400
|
||||
|
||||
df = algo_get_kline_data(stock_code, days=180)
|
||||
df = algo_get_kline_data(stock_code, days=120)
|
||||
if df is None or len(df) < 30:
|
||||
return jsonify({'error': 'K线数据不足'}), 400
|
||||
|
||||
|
||||
@@ -134,10 +134,16 @@ def compute_comprehensive_score(stock_code, stock_name, technical_score, df=None
|
||||
external_score = 0
|
||||
summaries = []
|
||||
|
||||
# ---- P0: 主力资金进出 ----
|
||||
# ---- P0: 主力资金进出(带30分钟缓存,与批量模式一致)----
|
||||
try:
|
||||
from services.fund_flow_analyzer import analyze_fund_flow
|
||||
fund_result = analyze_fund_flow(stock_code, days=5)
|
||||
now = time.time()
|
||||
cached_ff = _fund_flow_cache.get(stock_code)
|
||||
if cached_ff and (now - cached_ff[1]) < _FUND_FLOW_TTL:
|
||||
fund_result = cached_ff[0]
|
||||
else:
|
||||
from services.fund_flow_analyzer import analyze_fund_flow
|
||||
fund_result = analyze_fund_flow(stock_code, days=5)
|
||||
_fund_flow_cache[stock_code] = (fund_result, now)
|
||||
factors['fund_flow'] = fund_result
|
||||
external_score += fund_result.get('score', 0)
|
||||
all_reasons.extend(fund_result.get('reasons', []))
|
||||
|
||||
@@ -340,7 +340,9 @@ def detect_all_signals(df, lookback=5):
|
||||
if not np.issubdtype(df[col].dtype, np.floating):
|
||||
df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype(np.float64)
|
||||
|
||||
df = calc_all_indicators(df)
|
||||
# 若已有指标列则跳过重复计算(deep_analyze 等场景预计算过)
|
||||
if 'dif' not in df.columns or 'ma5' not in df.columns:
|
||||
df = calc_all_indicators(df)
|
||||
|
||||
all_signals = []
|
||||
all_signals.extend(detect_main_rising_wave(df, lookback))
|
||||
|
||||
@@ -172,7 +172,6 @@ def _get_kline_from_local_db(stock_code, days=120):
|
||||
conn = get_db()
|
||||
if not conn:
|
||||
return None
|
||||
conn.autocommit = True
|
||||
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
|
||||
@@ -7,11 +7,16 @@ from datetime import datetime, timedelta
|
||||
import traceback
|
||||
import json
|
||||
import os
|
||||
import time
|
||||
import threading
|
||||
from config import Config
|
||||
|
||||
|
||||
# ========== 股票名称缓存 ==========
|
||||
_stock_name_cache = {}
|
||||
_name_cache_dirty = False
|
||||
_name_cache_last_save = 0.0
|
||||
_name_cache_lock = threading.Lock()
|
||||
|
||||
|
||||
def _load_stock_name_cache():
|
||||
@@ -26,23 +31,51 @@ def _load_stock_name_cache():
|
||||
print(f"加载股票名称缓存失败: {e}")
|
||||
|
||||
|
||||
def _save_stock_name_cache():
|
||||
"""保存股票名称缓存到本地"""
|
||||
try:
|
||||
with open(Config.STOCK_NAME_CACHE_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump(_stock_name_cache, f, ensure_ascii=False, indent=2)
|
||||
except Exception as e:
|
||||
print(f"保存股票名称缓存失败: {e}")
|
||||
def _save_stock_name_cache(force=False):
|
||||
"""保存股票名称缓存到本地(延迟批量保存,60秒去抖)"""
|
||||
global _name_cache_dirty, _name_cache_last_save
|
||||
if not force and not _name_cache_dirty:
|
||||
return
|
||||
now = time.time()
|
||||
if not force and (now - _name_cache_last_save) < 60:
|
||||
return
|
||||
with _name_cache_lock:
|
||||
try:
|
||||
with open(Config.STOCK_NAME_CACHE_FILE, 'w', encoding='utf-8') as f:
|
||||
json.dump(_stock_name_cache, f, ensure_ascii=False, indent=2)
|
||||
_name_cache_dirty = False
|
||||
_name_cache_last_save = now
|
||||
except Exception as e:
|
||||
print(f"保存股票名称缓存失败: {e}")
|
||||
|
||||
|
||||
def get_stock_name(stock_code):
|
||||
"""获取股票名称 — 使用腾讯财经API"""
|
||||
global _stock_name_cache
|
||||
|
||||
"""获取股票名称 — 优先内存缓存 → DB stock_realtime_price → 腾讯财经API"""
|
||||
global _stock_name_cache, _name_cache_dirty
|
||||
|
||||
if stock_code in _stock_name_cache:
|
||||
return _stock_name_cache[stock_code]
|
||||
|
||||
# 腾讯财经API获取股票名称
|
||||
|
||||
# 优先从数据库 stock_realtime_price 表查(毫秒级)
|
||||
try:
|
||||
from db import get_db, put_db
|
||||
conn = get_db()
|
||||
if conn:
|
||||
try:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute("SELECT name FROM stock_realtime_price WHERE code = %s", (stock_code,))
|
||||
row = cur.fetchone()
|
||||
if row and row[0]:
|
||||
_stock_name_cache[stock_code] = row[0]
|
||||
_name_cache_dirty = True
|
||||
_save_stock_name_cache()
|
||||
return row[0]
|
||||
finally:
|
||||
put_db(conn)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# 回退到腾讯财经API
|
||||
try:
|
||||
import requests as _req
|
||||
tcode = ('sh' if stock_code.startswith('6') else 'bj' if stock_code.startswith(('8', '9')) else 'sz') + stock_code
|
||||
@@ -52,11 +85,12 @@ def get_stock_name(stock_code):
|
||||
_fields = _r.text.split('\"')[1].split('~')
|
||||
if len(_fields) > 2 and _fields[1]:
|
||||
_stock_name_cache[stock_code] = _fields[1]
|
||||
_name_cache_dirty = True
|
||||
_save_stock_name_cache()
|
||||
return _fields[1]
|
||||
except Exception as e:
|
||||
print(f"获取股票名称失败(腾讯): {e}")
|
||||
|
||||
|
||||
return None
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user