fix: 修复6个数据和代码问题

1. 修复 _generate_plain_summary position key 不匹配 (20d→d20, pct→range_pct) 2. 修复 mairui_api.py 日期解析不一致及 float(None) 崩溃风险 3. 修复 fund_flow_analyzer.py 麦蕊回退数据 close_price=0 导致量价背离误判 4. 修复 db_get_fund_flow_history 缺少完整字段和日期过滤 5. 修复 scheduler.py 连接池泄漏 (conn.close() → put_db(conn)) 6. 修复多处北交所股票代码映射缺失 (8/9开头→bj)
2026-07-22 07:20:44 +08:00
parent 9a5e24ccd6
commit 40ce519188
7 changed files with 84 additions and 37 deletions
@@ -552,7 +552,7 @@ def db_save_fundamental(code, data):

 # ========== 资金流向历史数据操作（数据库版） ==========

-def db_get_fund_flow_history(code):
+def db_get_fund_flow_history(code, limit=60):
    """获取股票的资金流向历史数据"""
    conn = get_db()
    if not conn:
@@ -564,11 +564,14 @@ def db_get_fund_flow_history(code):
            SELECT code, trade_date::text, close_price, change_pct,
                   main_net_inflow, main_net_inflow_pct,
                   super_net_inflow, super_net_inflow_pct,
-                   big_net_inflow, big_net_inflow_pct
+                   big_net_inflow, big_net_inflow_pct,
+                   mid_net_inflow, mid_net_inflow_pct,
+                   small_net_inflow, small_net_inflow_pct
            FROM stock_fund_flow_history 
            WHERE code = %s
            ORDER BY trade_date DESC
-        """, (code,))
+            LIMIT %s
+        """, (code, limit))
        rows = cur.fetchall()
        
        # 获取最新日期
@@ -61,7 +61,7 @@ def analyze():
        # 3. 获取实时价格补充到结果（优先腾讯API，兼容腾讯云）
        try:
            import requests as _req
-            _tcode = ('sh' if stock_code.startswith('6') else 'sz') + stock_code
+            _tcode = ('sh' if stock_code.startswith('6') else 'bj' if stock_code.startswith(('8', '9')) else 'sz') + stock_code
            _r = _req.get(f'http://qt.gtimg.cn/q={_tcode}', timeout=5,
                          headers={'Referer': 'https://finance.qq.com'})
            if _r.status_code == 200 and '\"' in _r.text:
@@ -155,6 +155,39 @@ def _get_fund_flow_from_mairui(stock_code, days=10):
                'small_net_inflow_pct': small_net_pct,
            })

+        # 从本地DB补充 close_price 和 change_pct（避免全为0导致量价背离误判）
+        from db import get_db, put_db
+        conn = get_db()
+        if conn:
+            try:
+                cur = conn.cursor()
+                date_list = [r['date'] for r in records if r['date']]
+                if date_list:
+                    cur.execute("""
+                        SELECT k.trade_date::text, k.close,
+                               CASE WHEN prev.close > 0
+                                    THEN ROUND((k.close - prev.close) / prev.close * 100, 2)
+                                    ELSE 0 END AS change_pct
+                        FROM stock_kline_daily k
+                        LEFT JOIN LATERAL (
+                            SELECT close FROM stock_kline_daily
+                            WHERE code = k.code AND trade_date < k.trade_date
+                            ORDER BY trade_date DESC LIMIT 1
+                        ) prev ON true
+                        WHERE k.code = %s AND k.trade_date::text = ANY(%s)
+                    """, (stock_code, date_list))
+                    price_map = {r[0]: {'close': float(r[1] or 0), 'change_pct': float(r[2] or 0)}
+                                 for r in cur.fetchall()}
+                    for r in records:
+                        info = price_map.get(r['date'])
+                        if info:
+                            r['close_price'] = info['close']
+                            r['change_pct'] = info['change_pct']
+            except Exception as e:
+                logger.warning(f"补充K线价格失败({stock_code}): {e}")
+            finally:
+                put_db(conn)
+
        # 按日期升序排列
        records.sort(key=lambda x: x['date'])
        logger.info(f"麦蕊API获取{stock_code}资金流向{len(records)}条")
@@ -245,17 +278,20 @@ def analyze_fund_flow(stock_code, days=5):
    # 出货：主力净流出但价格不跌（跌幅<2%）
    accumulation = False
    distribution = False
-    if total_main_inflow > 0:
-        price_changes = [r['change_pct'] for r in recent]
-        avg_price_change = sum(price_changes) / len(price_changes) if price_changes else 0
-        if avg_price_change < 2:
-            accumulation = True
+    # 检查是否有有效的价格数据（close_price 全为0说明数据不完整，跳过背离检测）
+    has_valid_price = any(r.get('close_price', 0) > 0 for r in recent)
+    if has_valid_price:
+        if total_main_inflow > 0:
+            price_changes = [r['change_pct'] for r in recent]
+            avg_price_change = sum(price_changes) / len(price_changes) if price_changes else 0
+            if avg_price_change < 2:
+                accumulation = True

-    if total_main_inflow < 0:
-        price_changes = [r['change_pct'] for r in recent]
-        avg_price_change = sum(price_changes) / len(price_changes) if price_changes else 0
-        if avg_price_change > -2:
-            distribution = True
+        if total_main_inflow < 0:
+            price_changes = [r['change_pct'] for r in recent]
+            avg_price_change = sum(price_changes) / len(price_changes) if price_changes else 0
+            if avg_price_change > -2:
+                distribution = True

    # 单日超大单突击
    big_surge = False
@@ -245,17 +245,21 @@ def get_fund_flow(stock_code, days=3):
        flow_data = []
        for item in data:
            # 计算主力净流入 = 主买大单+主买特大单 - 主卖大单-主卖特大单
-            main_buy = float(item.get('zmbddcje', 0)) + float(item.get('zmbtdcje', 0))
-            main_sell = float(item.get('zmsddcje', 0)) + float(item.get('zmstdcje', 0))
+            main_buy = float(item.get('zmbddcje', 0) or 0) + float(item.get('zmbtdcje', 0) or 0)
+            main_sell = float(item.get('zmsddcje', 0) or 0) + float(item.get('zmstdcje', 0) or 0)
            main_net = main_buy - main_sell
-            
+
+            # 日期解析：与 fund_flow_analyzer.py 保持一致，用字符串截取
+            t_str = str(item.get('t', ''))
+            date_str = t_str[:10] if t_str else ''
+
            flow_data.append({
-                'date': datetime.fromtimestamp(item.get('t', 0)).strftime('%Y-%m-%d') if item.get('t') else '',
+                'date': date_str,
                'main_net_inflow': main_net,
-                'super_buy': float(item.get('zmbtdcje', 0)),
-                'super_sell': float(item.get('zmstdcje', 0)),
-                'big_buy': float(item.get('zmbddcje', 0)),
-                'big_sell': float(item.get('zmsddcje', 0)),
+                'super_buy': float(item.get('zmbtdcje', 0) or 0),
+                'super_sell': float(item.get('zmstdcje', 0) or 0),
+                'big_buy': float(item.get('zmbddcje', 0) or 0),
+                'big_sell': float(item.get('zmsddcje', 0) or 0),
            })
        return {'success': True, 'data': flow_data}
    
@@ -101,7 +101,7 @@ def is_trading_time():

 def get_all_users():
    """获取所有启用自动交易的用户"""
-    from db import get_db
+    from db import get_db, put_db
    from psycopg2.extras import RealDictCursor
    
    conn = get_db()
@@ -121,7 +121,7 @@ def get_all_users():
        print(f"[定时任务] 获取用户列表失败: {e}")
        return []
    finally:
-        conn.close()
+        put_db(conn)


 # 自定义股票列表（100只精选股票）
@@ -168,7 +168,7 @@ def execute_auto_trade_for_user(user_id, trade_quantity=1000, scan_date=None):
    
    scan_date: 使用哪天的扫描数据, None则自动选择最近可用的
    """
-    from db import get_db
+    from db import get_db, put_db
    from psycopg2.extras import RealDictCursor

    print(f"[定时任务] 开始为用户{user_id}执行策略交易（统一推荐算法）...")
@@ -407,7 +407,7 @@ def execute_auto_trade_for_user(user_id, trade_quantity=1000, scan_date=None):
        traceback.print_exc()
        return {'error': str(e)}
    finally:
-        conn.close()
+        put_db(conn)


 def _get_latest_price(stock_code):
@@ -417,7 +417,7 @@ def _get_latest_price(stock_code):

 def update_positions_price_for_user(user_id):
    """更新用户持仓的当前价格（收盘时调用）— 使用腾讯财经API"""
-    from db import get_db
+    from db import get_db, put_db
    from psycopg2.extras import RealDictCursor
    
    conn = get_db()
@@ -443,6 +443,8 @@ def update_positions_price_for_user(user_id):
                for c in codes:
                    if c.startswith('6'):
                        tencent_codes.append(f'sh{c}')
+                    elif c.startswith('8') or c.startswith('9'):
+                        tencent_codes.append(f'bj{c}')
                    else:
                        tencent_codes.append(f'sz{c}')
                _r = _req.get(f'http://qt.gtimg.cn/q={",".join(tencent_codes)}',
@@ -492,7 +494,7 @@ def update_positions_price_for_user(user_id):
        conn.rollback()
        print(f"[定时任务] 更新用户{user_id}持仓价格失败: {e}")
    finally:
-        conn.close()
+        put_db(conn)


 def job_morning_trade():
@@ -514,11 +516,11 @@ def job_morning_trade():
        # 尝试使用智能引擎
        try:
            from services.smart_trade_engine import execute_smart_trade
-            from db import get_db
+            from db import get_db, put_db
            conn = get_db()
            if conn:
                result = execute_smart_trade(conn, user_id, scan_date=None)
-                conn.close()
+                put_db(conn)
                if result.get('success'):
                    print(f"[定时任务] 用户{user_id} 智能引擎执行成功 "
                          f"(算法:{result.get('algo','?')}, 信号:{result.get('signals',0)})")
@@ -553,11 +555,11 @@ def job_afternoon_trade():
        # 尝试使用智能引擎
        try:
            from services.smart_trade_engine import execute_smart_trade
-            from db import get_db
+            from db import get_db, put_db
            conn = get_db()
            if conn:
                result = execute_smart_trade(conn, user_id, scan_date=today)
-                conn.close()
+                put_db(conn)
                if result.get('success'):
                    print(f"[定时任务] 用户{user_id} 午后智能引擎执行成功")
                    continue
@@ -645,7 +647,7 @@ def trigger_afternoon_trade():

 def trigger_closing_update():
    """手动触发收盘更新（仅更新持仓价格）"""
-    from db import get_db
+    from db import get_db, put_db
    if not is_trading_day():
        print("[定时任务] 今天不是交易日，跳过")
        return
@@ -962,8 +962,8 @@ def _generate_plain_summary(price, change_pct, ma_trend, position, supports,
    parts.append(trend_desc + '。')

    # 2. 价格位置（用大白话）
-    pos_20 = position.get('20d', {})
-    pct_20 = pos_20.get('pct', 50)
+    pos_20 = position.get('d20', {})
+    pct_20 = pos_20.get('range_pct', 50)
    if pct_20 > 80:
        parts.append(f'当前股价处于近20天的高位区间（{pct_20:.0f}%位置），已经涨了不少，追高要小心。')
    elif pct_20 > 50:
@@ -45,7 +45,7 @@ def get_stock_name(stock_code):
    # 腾讯财经API获取股票名称
    try:
        import requests as _req
-        tcode = ('sh' if stock_code.startswith('6') else 'sz') + stock_code
+        tcode = ('sh' if stock_code.startswith('6') else 'bj' if stock_code.startswith(('8', '9')) else 'sz') + stock_code
        _r = _req.get(f'http://qt.gtimg.cn/q={tcode}', timeout=5,
                      headers={'Referer': 'https://finance.qq.com'})
        if _r.status_code == 200 and '\"' in _r.text:
@@ -119,6 +119,8 @@ def get_stock_fund_flow(stock_code, start_date, end_date, force_refresh=False):
            market = 'sh'
        elif stock_code.startswith('0') or stock_code.startswith('3'):
            market = 'sz'
+        elif stock_code.startswith('8') or stock_code.startswith('9'):
+            market = 'bj'
        else:
            return None, None, "无法识别股票代码所属市场"
        
@@ -313,7 +315,7 @@ def get_realtime_price(stock_code):
    # 备用方案2：使用腾讯财经API（腾讯云可用）
    try:
        import requests as _req
-        tcode = ('sh' if stock_code.startswith('6') else 'sz') + stock_code
+        tcode = ('sh' if stock_code.startswith('6') else 'bj' if stock_code.startswith(('8', '9')) else 'sz') + stock_code
        _r = _req.get(f'http://qt.gtimg.cn/q={tcode}', timeout=5,
                      headers={'Referer': 'https://finance.qq.com'})
        if _r.status_code == 200 and '\"' in _r.text: