更新Company页面的UI为FUI风格

app.py

@@ -371,6 +371,199 @@ def wechat_session_exists(state):
         print(f"❌ Redis 检查 wechat session 失败: {e}")
         return False
 # ============ 微信登录 Session 管理结束 ============
+
+# ============ 股票数据 Redis 缓存（股票名称 + 前收盘价） ============
+STOCK_NAME_PREFIX = "vf:stock:name:"           # 股票名称缓存前缀
+STOCK_NAME_EXPIRE = 86400                       # 股票名称缓存24小时
+PREV_CLOSE_PREFIX = "vf:stock:prev_close:"     # 前收盘价缓存前缀
+PREV_CLOSE_EXPIRE = 86400                       # 前收盘价缓存24小时（当日有效）
+
+
+def get_cached_stock_names(base_codes):
+    """
+    批量获取股票名称（优先从 Redis 缓存读取）
+    :param base_codes: 股票代码列表（不带后缀，如 ['600000', '000001']）
+    :return: dict {code: name}
+    """
+    if not base_codes:
+        return {}
+
+    result = {}
+    missing_codes = []
+
+    try:
+        # 批量从 Redis 获取
+        pipe = redis_client.pipeline()
+        for code in base_codes:
+            pipe.get(f"{STOCK_NAME_PREFIX}{code}")
+        cached_values = pipe.execute()
+
+        for code, cached_name in zip(base_codes, cached_values):
+            if cached_name:
+                result[code] = cached_name
+            else:
+                missing_codes.append(code)
+    except Exception as e:
+        print(f"⚠️ Redis 批量获取股票名称失败: {e}，降级为数据库查询")
+        missing_codes = base_codes
+
+    # 从数据库查询缺失的股票名称
+    if missing_codes:
+        try:
+            with engine.connect() as conn:
+                placeholders = ','.join([f':code{i}' for i in range(len(missing_codes))])
+                params = {f'code{i}': code for i, code in enumerate(missing_codes)}
+                db_result = conn.execute(text(
+                    f"SELECT SECCODE, SECNAME FROM ea_stocklist WHERE SECCODE IN ({placeholders})"
+                ), params).fetchall()
+
+                # 写入 Redis 缓存
+                pipe = redis_client.pipeline()
+                for row in db_result:
+                    code, name = row[0], row[1]
+                    result[code] = name
+                    pipe.setex(f"{STOCK_NAME_PREFIX}{code}", STOCK_NAME_EXPIRE, name)
+
+                try:
+                    pipe.execute()
+                except Exception as e:
+                    print(f"⚠️ Redis 缓存股票名称失败: {e}")
+        except Exception as e:
+            print(f"❌ 数据库查询股票名称失败: {e}")
+
+    return result
+
+
+def get_cached_prev_close(base_codes, trade_date_str):
+    """
+    批量获取前收盘价（优先从 Redis 缓存读取）
+    :param base_codes: 股票代码列表（不带后缀，如 ['600000', '000001']）
+    :param trade_date_str: 交易日期字符串（格式 YYYYMMDD）
+    :return: dict {code: close_price}
+    """
+    if not base_codes or not trade_date_str:
+        return {}
+
+    result = {}
+    missing_codes = []
+
+    try:
+        # 批量从 Redis 获取（缓存键包含日期，确保不会跨日混用）
+        pipe = redis_client.pipeline()
+        for code in base_codes:
+            pipe.get(f"{PREV_CLOSE_PREFIX}{trade_date_str}:{code}")
+        cached_values = pipe.execute()
+
+        for code, cached_price in zip(base_codes, cached_values):
+            if cached_price:
+                result[code] = float(cached_price)
+            else:
+                missing_codes.append(code)
+    except Exception as e:
+        print(f"⚠️ Redis 批量获取前收盘价失败: {e}，降级为数据库查询")
+        missing_codes = base_codes
+
+    # 从数据库查询缺失的前收盘价
+    if missing_codes:
+        try:
+            with engine.connect() as conn:
+                placeholders = ','.join([f':code{i}' for i in range(len(missing_codes))])
+                params = {f'code{i}': code for i, code in enumerate(missing_codes)}
+                params['trade_date'] = trade_date_str
+                db_result = conn.execute(text(f"""
+                    SELECT SECCODE, F007N as close_price
+                    FROM ea_trade
+                    WHERE SECCODE IN ({placeholders})
+                      AND TRADEDATE = :trade_date
+                      AND F007N > 0
+                """), params).fetchall()
+
+                # 写入 Redis 缓存
+                pipe = redis_client.pipeline()
+                for row in db_result:
+                    code, close_price = row[0], float(row[1]) if row[1] else None
+                    if close_price:
+                        result[code] = close_price
+                        pipe.setex(f"{PREV_CLOSE_PREFIX}{trade_date_str}:{code}", PREV_CLOSE_EXPIRE, str(close_price))
+
+                try:
+                    pipe.execute()
+                except Exception as e:
+                    print(f"⚠️ Redis 缓存前收盘价失败: {e}")
+        except Exception as e:
+            print(f"❌ 数据库查询前收盘价失败: {e}")
+
+    return result
+
+
+def preload_stock_cache():
+    """
+    预热股票缓存（定时任务，每天 9:25 执行）
+    - 批量加载所有股票名称
+    - 批量加载前一交易日收盘价
+    """
+    from datetime import datetime, timedelta
+    print(f"[缓存预热] 开始预热股票缓存... {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+
+    try:
+        # 1. 预热股票名称（全量加载）
+        with engine.connect() as conn:
+            result = conn.execute(text("SELECT SECCODE, SECNAME FROM ea_stocklist")).fetchall()
+            pipe = redis_client.pipeline()
+            count = 0
+            for row in result:
+                code, name = row[0], row[1]
+                if code and name:
+                    pipe.setex(f"{STOCK_NAME_PREFIX}{code}", STOCK_NAME_EXPIRE, name)
+                    count += 1
+            pipe.execute()
+            print(f"[缓存预热] 股票名称: {count} 条已加载到 Redis")
+
+        # 2. 预热前收盘价（获取前一交易日）
+        # 使用全局 trading_days 获取前一交易日
+        today = datetime.now().date()
+        today_str = today.strftime('%Y-%m-%d')
+
+        prev_trading_day = None
+        if 'trading_days' in globals() and trading_days:
+            # 找到今天之前最近的交易日
+            for td in reversed(trading_days):
+                if td < today_str:
+                    prev_trading_day = td
+                    break
+
+        if prev_trading_day:
+            prev_date_str = prev_trading_day.replace('-', '')  # YYYYMMDD 格式
+            with engine.connect() as conn:
+                result = conn.execute(text("""
+                    SELECT SECCODE, F007N as close_price
+                    FROM ea_trade
+                    WHERE TRADEDATE = :trade_date AND F007N > 0
+                """), {'trade_date': prev_date_str}).fetchall()
+
+                pipe = redis_client.pipeline()
+                count = 0
+                for row in result:
+                    code, close_price = row[0], row[1]
+                    if code and close_price:
+                        pipe.setex(f"{PREV_CLOSE_PREFIX}{prev_date_str}:{code}", PREV_CLOSE_EXPIRE, str(close_price))
+                        count += 1
+                pipe.execute()
+                print(f"[缓存预热] 前收盘价({prev_trading_day}): {count} 条已加载到 Redis")
+        else:
+            print(f"[缓存预热] 未找到前一交易日，跳过前收盘价预热")
+
+        print(f"[缓存预热] 预热完成 ✅ {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+
+    except Exception as e:
+        print(f"[缓存预热] 预热失败 ❌: {e}")
+        import traceback
+        traceback.print_exc()
+
+
+print(f"📦 股票缓存: Redis, 名称过期 {STOCK_NAME_EXPIRE}秒, 收盘价过期 {PREV_CLOSE_EXPIRE}秒")
+# ============ 股票数据 Redis 缓存结束 ============
+
 # 腾讯云短信配置
 SMS_SECRET_ID = 'AKID2we9TacdTAhCjCSYTErHVimeJo9Yr00s'
 SMS_SECRET_KEY = 'pMlBWijlkgT9fz5ziEXdWEnAPTJzRfkf'
@@ -7310,21 +7503,9 @@ def get_stock_quotes():
 
         current_time = datetime.now()
 
-        # ==================== 查询股票名称（直接查 MySQL） ====================
+        # ==================== 查询股票名称（使用 Redis 缓存） ====================
-        stock_names = {}
         base_codes = list(set([code.split('.')[0] for code in codes]))
-
+        stock_names = get_cached_stock_names(base_codes)
-        if base_codes:
-            with engine.connect() as conn:
-                placeholders = ','.join([f':code{i}' for i in range(len(base_codes))])
-                params = {f'code{i}': code for i, code in enumerate(base_codes)}
-                result = conn.execute(text(
-                    f"SELECT SECCODE, SECNAME FROM ea_stocklist WHERE SECCODE IN ({placeholders})"
-                ), params).fetchall()
-
-                for row in result:
-                    base_code, name = row[0], row[1]
-                    stock_names[base_code] = name
 
         # 构建完整的名称映射
         full_stock_names = {}
@@ -7355,34 +7536,17 @@ def get_stock_quotes():
         # 初始化 ClickHouse 客户端
         client = get_clickhouse_client()
 
-        # ==================== 查询前一交易日收盘价（直接查 MySQL） ====================
+        # ==================== 查询前一交易日收盘价（使用 Redis 缓存） ====================
         try:
             prev_close_map = {}
             if prev_trading_day:
                 # ea_trade 表的 TRADEDATE 格式是 YYYYMMDD（无连字符）
                 prev_day_str = prev_trading_day.strftime('%Y%m%d') if hasattr(prev_trading_day, 'strftime') else str(prev_trading_day).replace('-', '')
                 base_codes = list(set([code.split('.')[0] for code in codes]))
-                base_close_map = {}
 
-                # 直接从 MySQL 批量查询
+                # 使用 Redis 缓存获取前收盘价
-                with engine.connect() as conn:
+                base_close_map = get_cached_prev_close(base_codes, prev_day_str)
-                    placeholders = ','.join([f':code{i}' for i in range(len(base_codes))])
+                print(f"前一交易日({prev_day_str})收盘价: 获取到 {len(base_close_map)} 条（Redis缓存）")
-                    params = {f'code{i}': code for i, code in enumerate(base_codes)}
-                    params['trade_date'] = prev_day_str
-
-                    prev_close_result = conn.execute(text(f"""
-                        SELECT SECCODE, F007N as close_price
-                        FROM ea_trade
-                        WHERE SECCODE IN ({placeholders})
-                          AND TRADEDATE = :trade_date
-                    """), params).fetchall()
-
-                    for row in prev_close_result:
-                        base_code, close_price = row[0], row[1]
-                        close_val = float(close_price) if close_price else None
-                        base_close_map[base_code] = close_val
-
-                    print(f"前一交易日({prev_day_str})收盘价: 查询到 {len(prev_close_result)} 条")
 
                 # 为每个标准化代码分配收盘价
                 for norm_code in normalized_codes:
@@ -8142,18 +8306,9 @@ def get_batch_kline_data():
 
         client = get_clickhouse_client()
 
-        # 批量获取股票名称
+        # 批量获取股票名称（使用 Redis 缓存）
-        stock_names = {}
+        base_codes = list(set([code.split('.')[0] for code in codes]))
-        with engine.connect() as conn:
+        stock_names = get_cached_stock_names(base_codes)
-            base_codes = list(set([code.split('.')[0] for code in codes]))
-            if base_codes:
-                placeholders = ','.join([f':code{i}' for i in range(len(base_codes))])
-                params = {f'code{i}': code for i, code in enumerate(base_codes)}
-                result = conn.execute(text(
-                    f"SELECT SECCODE, SECNAME FROM ea_stocklist WHERE SECCODE IN ({placeholders})"
-                ), params).fetchall()
-                for row in result:
-                    stock_names[row[0]] = row[1]
 
         # 确定目标交易日和涨跌幅基准日（处理跨周末场景）
         # - 周五15:00后到周一15:00前，分时图显示周一行情，涨跌幅基于周五收盘价
@@ -8172,24 +8327,14 @@ def get_batch_kline_data():
         results = {}
 
         if chart_type == 'timeline':
-            # 批量获取前收盘价（从 MySQL ea_trade 表）
+            # 批量获取前收盘价（使用 Redis 缓存）
             # 使用 prev_trading_day 作为基准日期（处理跨周末场景）
             prev_close_map = {}
             if prev_trading_day:
                 prev_date_str = prev_trading_day.strftime('%Y%m%d')
-                with engine.connect() as conn:
+                base_codes = list(set([code.split('.')[0] for code in codes]))
-                    base_codes = list(set([code.split('.')[0] for code in codes]))
+                prev_close_map = get_cached_prev_close(base_codes, prev_date_str)
-                    if base_codes:
+                print(f"分时图基准日期: {prev_trading_day}, 获取到 {len(prev_close_map)} 条前收盘价（Redis缓存）")
-                        placeholders = ','.join([f':code{i}' for i in range(len(base_codes))])
-                        params = {f'code{i}': code for i, code in enumerate(base_codes)}
-                        params['trade_date'] = prev_date_str
-                        result = conn.execute(text(f"""
-                            SELECT SECCODE, F007N FROM ea_trade
-                            WHERE SECCODE IN ({placeholders}) AND TRADEDATE = :trade_date AND F007N > 0
-                        """), params).fetchall()
-                        for row in result:
-                            prev_close_map[row[0]] = float(row[1])
-                print(f"分时图基准日期: {prev_trading_day}, 查询到 {len(prev_close_map)} 条前收盘价")
 
             # 批量查询分时数据（使用标准化代码查询 ClickHouse）
             batch_data = client.execute("""
@@ -11893,6 +12038,18 @@ def initialize_event_polling():
                 name='检查新事件并推送',
                 replace_existing=True
             )
+
+            # 每天 9:25 预热股票缓存（开盘前 5 分钟）
+            from apscheduler.triggers.cron import CronTrigger
+            scheduler.add_job(
+                func=preload_stock_cache,
+                trigger=CronTrigger(hour=9, minute=25),
+                id='preload_stock_cache',
+                name='预热股票缓存（股票名称+前收盘价）',
+                replace_existing=True
+            )
+            print(f'[缓存] 已添加定时任务: 每天 9:25 预热股票缓存')
+
             scheduler.start()
             print(f'[轮询] APScheduler 调度器已启动 (PID: {os.getpid()})，每 30 秒检查一次新事件')
 

get_related_chg.py

@@ -245,10 +245,17 @@ def update_event_statistics(start_date=None, end_date=None, force_update=False,
         debug_mode: 是否开启调试模式
     """
     try:
+        print("[DEBUG] 开始 update_event_statistics")
+        print(f"[DEBUG] 参数: start_date={start_date}, end_date={end_date}, force_update={force_update}")
+
         mysql_engine = get_mysql_engine()
+        print("[DEBUG] MySQL 引擎创建成功")
+
         ch_client = get_clickhouse_client()
+        print("[DEBUG] ClickHouse 客户端创建成功")
 
         with mysql_engine.connect() as mysql_conn:
+            print("[DEBUG] MySQL 连接已建立")
             # 构建SQL查询
             query = """
                     SELECT e.id, \
@@ -284,8 +291,12 @@ def update_event_statistics(start_date=None, end_date=None, force_update=False,
                 ORDER BY e.created_at DESC
             """
 
+            print(f"[DEBUG] 执行查询SQL:\n{query}")
+            print(f"[DEBUG] 查询参数: {params}")
+
             events = mysql_conn.execute(text(query), params).fetchall()
 
+            print(f"[DEBUG] 查询返回 {len(events)} 条事件记录")
             print(f"Found {len(events)} events to update (force_update={force_update})")
             if debug_mode and len(events) > 0:
                 print(f"Date range: {events[-1][1]} to {events[0][1]}")
@@ -324,6 +335,11 @@ def update_event_statistics(start_date=None, end_date=None, force_update=False,
                             "week_chg": week_change,
                             "event_id": event_id
                         })
+                        if idx <= 5:  # 前5条显示详情
+                            print(f"[DEBUG] 事件 {event_id}: avg={avg_change}, max={max_change}, week={week_change}")
+                    else:
+                        if idx <= 5:
+                            print(f"[DEBUG] 事件 {event_id}: 计算结果全为None，跳过")
 
                     # 每处理10个事件打印一次进度
                     if idx % 10 == 0:
@@ -337,16 +353,29 @@ def update_event_statistics(start_date=None, end_date=None, force_update=False,
                     continue
 
             # 批量更新MySQL
+            print(f"\n[DEBUG] ====== 准备写入数据库 ======")
+            print(f"[DEBUG] update_data 长度: {len(update_data)}")
             if update_data:
-                mysql_conn.execute(text("""
+                print(f"[DEBUG] 前3条待更新数据: {update_data[:3]}")
+                print(f"[DEBUG] 执行 UPDATE 语句...")
+
+                result = mysql_conn.execute(text("""
                                         UPDATE event
                                         SET related_avg_chg  = :avg_chg,
                                             related_max_chg  = :max_chg,
                                             related_week_chg = :week_chg
                                         WHERE id = :event_id
                                         """), update_data)
-                # SQLAlchemy 1.4 的 connect() 会在上下文管理器退出时自动提交
+                print(f"[DEBUG] UPDATE 执行完成, rowcount={result.rowcount}")
+
+                # 关键：显式提交事务！SQLAlchemy 2.0 需要手动 commit
+                print("[DEBUG] 准备提交事务 (commit)...")
+                mysql_conn.commit()
+                print("[DEBUG] 事务已提交!")
+
                 print(f"Successfully updated {len(update_data)} events")
+            else:
+                print("[DEBUG] update_data 为空，没有数据需要更新!")
 
     except Exception as e:
         print(f"Error in update_event_statistics: {str(e)}")