新增获取历史二板数据脚本

# Conflicts:
#	cal_date.py
#	daily.py
#	limit_up.py

basic.py

@@ -1,9 +1,10 @@
 import tushare as ts
-from sqlalchemy import create_engine
+
+from stock_config import TUSHARE_TOKEN, get_engine
 
 # 1. 配置参数
 # 在 Tushare 官网个人主页获取你的 token
-TOKEN = '0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93'
+TOKEN = TUSHARE_TOKEN
 # 数据库连接配置 (格式: mysql+pymysql://用户名:密码@主机地址:端口/数据库名)
 # DB_CONN_STR = 'mysql+pymysql://yangfan:aA%8023321088@8.159.129.156:10836/level?charset=utf8'
 
@@ -25,18 +26,12 @@ def sync_stock_basic():
 
     if df.empty:
         print("未获取到数据，请检查接口权限或Token设置。")
-        return
+        return False
 
     # 4. 写入数据库
     try:
         # 创建数据库引擎
-        engine = create_engine(
+        engine = get_engine()
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
 
         # 将数据写入数据库表 'stock_basic'
         # if_exists='replace': 如果表存在则替换（适合全量更新基础信息）
@@ -44,9 +39,11 @@ def sync_stock_basic():
         df.to_sql('stock_basic', engine, if_exists='replace', index=False)
 
         print(f"成功将 {len(df)} 条股票基础信息写入数据库表 [stock_basic]。")
+        return True
 
     except Exception as e:
         print(f"写入数据库失败: {e}")
+        return False
 
 
 if __name__ == "__main__":

cal_date.py

@@ -1,7 +1,10 @@
 import tushare as ts
-from sqlalchemy import create_engine, text
+from sqlalchemy import text
 from datetime import datetime
 import sys
+import argparse
+
+from stock_config import TUSHARE_TOKEN, get_engine as create_db_engine
 
 # =======================
 # 1. 日期校验（YYYYMMDD）
@@ -36,14 +39,7 @@ def validate_dates(start_str, end_str):
 # 2. 创建数据库连接（解决 % 编码问题）
 # =======================
 def get_engine():
-    engine = create_engine(
+    return create_db_engine()
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
-    return engine
 
 
 # =======================
@@ -92,11 +88,11 @@ def run(start_date, end_date, token):
         )
     except Exception as e:
         print(f"❌ 获取数据失败: {e}")
-        return
+        return False
 
     if df is None or df.empty:
         print("⚠️ 无数据")
-        return
+        return False
 
     records = []
     for _, row in df.iterrows():
@@ -126,15 +122,16 @@ def run(start_date, end_date, token):
         conn.execute(sql, records)
 
     print(f"✅ 写入完成，共 {len(records)} 条")
+    return True
 
 
 # =======================
 # 5. 程序入口（只改这里）
 # =======================
 if __name__ == "__main__":
-    START_DATE = "20240427"
+    parser = argparse.ArgumentParser(description="同步交易日历")
-    END_DATE   = "20260427"
+    parser.add_argument("--start-date", default="20260513", help="开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD")
+    args = parser.parse_args()
 
-    TUSHARE_TOKEN = "0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93"
+    run(args.start_date, args.end_date, TUSHARE_TOKEN)
-
-    run(START_DATE, END_DATE, TUSHARE_TOKEN)

daily.py

@@ -1,30 +1,31 @@
 import tushare as ts
 import pymysql
+import pandas as pd
 from datetime import datetime
 import time
+import argparse
 
-# ===== 配置 =====
+from stock_config import DB_CONFIG, TUSHARE_TOKEN
-TUSHARE_TOKEN = "0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93"
-
-DB_CONFIG = {
-    "host": "8.159.129.156",
-    "port": 10836,
-    "user": "yangfan",
-    "password": "aA%8023321088",
-    "database": "level",
-    "charset": "utf8mb4"
-}
 
 # ===== 初始化 =====
 ts.set_token(TUSHARE_TOKEN)
 pro = ts.pro_api()
 
-conn = pymysql.connect(**DB_CONFIG)
+conn = None
-cursor = conn.cursor()
+cursor = None
+
+
+def get_connection():
+    global conn, cursor
+    if conn is None or not conn.open:
+        conn = pymysql.connect(**DB_CONFIG)
+        cursor = conn.cursor()
+    return conn, cursor
 
 
 # ===== 建表 =====
 def create_tables():
+    db_conn, db_cursor = get_connection()
     sql = """
     CREATE TABLE IF NOT EXISTS stock_daily (
         id BIGINT AUTO_INCREMENT PRIMARY KEY,
@@ -43,8 +44,8 @@ def create_tables():
         UNIQUE KEY uniq_code_date (ts_code, trade_date)
     );
     """
-    cursor.execute(sql)
+    db_cursor.execute(sql)
-    conn.commit()
+    db_conn.commit()
     print("表检查/创建完成")
 
 
@@ -53,8 +54,17 @@ def str_to_date(date_str):
     return datetime.strptime(date_str, "%Y%m%d").date()
 
 
+def mysql_value(value):
+    if pd.isna(value):
+        return None
+    if hasattr(value, "item"):
+        return value.item()
+    return value
+
+
 # ===== 获取交易日 =====
 def get_trade_dates(start_date, end_date):
+    _, db_cursor = get_connection()
     sql = """
     SELECT cal_date 
     FROM trade_calendar
@@ -62,14 +72,15 @@ def get_trade_dates(start_date, end_date):
       AND is_open = 1
     ORDER BY cal_date
     """
-    cursor.execute(sql, (str_to_date(start_date), str_to_date(end_date)))
+    db_cursor.execute(sql, (str_to_date(start_date), str_to_date(end_date)))
-    results = cursor.fetchall()
+    results = db_cursor.fetchall()
 
     return [row[0].strftime("%Y%m%d") for row in results]
 
 
 # ===== 插入数据 =====
 def insert_data(df):
+    db_conn, db_cursor = get_connection()
     if df is None or df.empty:
         return 0, 0
 
@@ -88,21 +99,21 @@ def insert_data(df):
             ts_code,
             symbol,
             str_to_date(row['trade_date']),  # 保持 DATE 类型
-            row['open'],
+            mysql_value(row['open']),
-            row['high'],
+            mysql_value(row['high']),
-            row['low'],
+            mysql_value(row['low']),
-            row['close'],
+            mysql_value(row['close']),
-            row['pre_close'],
+            mysql_value(row['pre_close']),
-            row['change'],
+            mysql_value(row['change']),
-            row['pct_chg'],
+            mysql_value(row['pct_chg']),
-            row['vol'],
+            mysql_value(row['vol']),
-            row['amount']
+            mysql_value(row['amount'])
         ))
 
-    cursor.executemany(sql, data)
+    db_cursor.executemany(sql, data)
-    conn.commit()
+    db_conn.commit()
 
-    inserted = cursor.rowcount
+    inserted = db_cursor.rowcount
     skipped = len(data) - inserted
 
     return inserted, skipped
@@ -140,11 +151,14 @@ def run(start_date, end_date):
     print("\n====== 汇总 ======")
     print(f"总新增: {total_inserted}")
     print(f"总跳过: {total_skipped}")
+    return True
 
 
 # ===== 输入参数 =====
 if __name__ == "__main__":
-    START_DATE = "20260427"
+    parser = argparse.ArgumentParser(description="同步日行情")
-    END_DATE = "20260427"
+    parser.add_argument("--start-date", default="20260513", help="开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD")
+    args = parser.parse_args()
 
-    run(START_DATE, END_DATE)
+    run(args.start_date, args.end_date)

limit_list_d.py

@@ -0,0 +1,283 @@
+import time
+from collections import deque
+from datetime import date, datetime
+
+import pandas as pd
+import tushare as ts
+from sqlalchemy import text
+from sqlalchemy.exc import SQLAlchemyError
+
+from stock_config import TUSHARE_TOKEN, get_engine
+
+
+TABLE_NAME = "limit_list_d"
+LIMIT_TYPES = ("U", "D", "Z")
+TUSHARE_START_DATE = date(2020, 1, 1)
+MAX_CALLS_PER_MINUTE = 200
+FIELDS = (
+    "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,"
+    "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+    "open_times,up_stat,limit_times,limit"
+)
+
+
+class MinuteRateLimiter:
+    def __init__(self, max_calls):
+        self.max_calls = max_calls
+        self.call_times = deque()
+
+    def wait(self):
+        now = time.monotonic()
+        while self.call_times and now - self.call_times[0] >= 60:
+            self.call_times.popleft()
+        if len(self.call_times) >= self.max_calls:
+            sleep_seconds = 60 - (now - self.call_times[0]) + 0.1
+            print(f"⏳ Tushare 限流等待 {sleep_seconds:.1f} 秒。")
+            time.sleep(sleep_seconds)
+            now = time.monotonic()
+            while self.call_times and now - self.call_times[0] >= 60:
+                self.call_times.popleft()
+        self.call_times.append(time.monotonic())
+
+
+def parse_date(value):
+    value = str(value).strip()
+    for fmt in ("%Y%m%d", "%Y-%m-%d"):
+        try:
+            return datetime.strptime(value, fmt).date()
+        except ValueError:
+            pass
+    raise ValueError(f"日期格式错误: {value}，应为 YYYYMMDD 或 YYYY-MM-DD")
+
+
+def to_tushare_date(value):
+    return value.strftime("%Y%m%d")
+
+
+def init_db(engine):
+    sql = f"""
+    CREATE TABLE IF NOT EXISTS `{TABLE_NAME}` (
+        `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
+        `trade_date` DATE NOT NULL COMMENT '交易日期',
+        `ts_code` VARCHAR(20) NOT NULL COMMENT '股票代码',
+        `industry` VARCHAR(100) COMMENT '所属行业',
+        `name` VARCHAR(50) COMMENT '股票名称',
+        `close` DECIMAL(16, 4) COMMENT '收盘价',
+        `pct_chg` DECIMAL(10, 4) COMMENT '涨跌幅',
+        `amount` DECIMAL(24, 4) COMMENT '成交额',
+        `limit_amount` DECIMAL(24, 4) COMMENT '板上成交金额',
+        `float_mv` DECIMAL(24, 4) COMMENT '流通市值',
+        `total_mv` DECIMAL(24, 4) COMMENT '总市值',
+        `turnover_ratio` DECIMAL(10, 4) COMMENT '换手率',
+        `fd_amount` DECIMAL(24, 4) COMMENT '封单金额',
+        `first_time` VARCHAR(20) COMMENT '首次封板时间',
+        `last_time` VARCHAR(20) COMMENT '最后封板时间',
+        `open_times` INT COMMENT '打开次数',
+        `up_stat` VARCHAR(50) COMMENT '涨停统计',
+        `limit_times` INT COMMENT '连板数',
+        `limit` VARCHAR(1) NOT NULL COMMENT 'D跌停 U涨停 Z炸板',
+        UNIQUE KEY `uniq_trade_code_limit` (`trade_date`, `ts_code`, `limit`),
+        KEY `idx_trade_date` (`trade_date`),
+        KEY `idx_ts_code` (`ts_code`),
+        KEY `idx_limit` (`limit`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
+    """
+    with engine.begin() as conn:
+        conn.execute(text(sql))
+    print(f"✅ 表 {TABLE_NAME} 检查/创建完成。")
+
+
+def get_trade_dates(engine, start_date, end_date):
+    try:
+        query = text(
+            """
+            SELECT cal_date
+            FROM trade_calendar
+            WHERE cal_date BETWEEN :start_date AND :end_date
+              AND is_open = 1
+            ORDER BY cal_date
+            """
+        )
+        df = pd.read_sql(
+            query,
+            engine,
+            params={"start_date": start_date, "end_date": end_date},
+        )
+        if not df.empty:
+            return [pd.to_datetime(value).date() for value in df["cal_date"]]
+    except SQLAlchemyError as exc:
+        print(f"⚠️ 读取 trade_calendar 失败，将按自然日尝试: {exc}")
+
+    return [value.date() for value in pd.date_range(start_date, end_date, freq="D")]
+
+
+def clean_text(value):
+    if pd.isna(value):
+        return None
+    return str(value).strip()
+
+
+def normalize_data(df):
+    columns = [
+        "trade_date",
+        "ts_code",
+        "industry",
+        "name",
+        "close",
+        "pct_chg",
+        "amount",
+        "limit_amount",
+        "float_mv",
+        "total_mv",
+        "turnover_ratio",
+        "fd_amount",
+        "first_time",
+        "last_time",
+        "open_times",
+        "up_stat",
+        "limit_times",
+        "limit",
+    ]
+    for column in columns:
+        if column not in df.columns:
+            df[column] = None
+
+    result = df[columns].copy()
+    result["trade_date"] = pd.to_datetime(
+        result["trade_date"], format="%Y%m%d", errors="coerce"
+    ).dt.date
+
+    for column in [
+        "close",
+        "pct_chg",
+        "amount",
+        "limit_amount",
+        "float_mv",
+        "total_mv",
+        "turnover_ratio",
+        "fd_amount",
+        "open_times",
+        "limit_times",
+    ]:
+        result[column] = pd.to_numeric(result[column], errors="coerce")
+
+    for column in [
+        "ts_code",
+        "industry",
+        "name",
+        "first_time",
+        "last_time",
+        "up_stat",
+        "limit",
+    ]:
+        result[column] = result[column].map(clean_text)
+
+    result = result.dropna(subset=["trade_date", "ts_code", "limit"])
+    return result.drop_duplicates(
+        subset=["trade_date", "ts_code", "limit"],
+        keep="first",
+    )
+
+
+def fetch_limit_list_d(trade_dates):
+    if not trade_dates:
+        return pd.DataFrame()
+
+    ts.set_token(TUSHARE_TOKEN)
+    pro = ts.pro_api()
+
+    frames = []
+    rate_limiter = MinuteRateLimiter(MAX_CALLS_PER_MINUTE)
+    total_calls = len(trade_dates) * len(LIMIT_TYPES)
+    current_call = 0
+
+    print(
+        f"🔎 开始获取 Tushare limit_list_d：{len(trade_dates)} 个交易日，"
+        f"{len(LIMIT_TYPES)} 类 limit_type，共 {total_calls} 次调用。"
+    )
+    for trade_date in trade_dates:
+        tushare_date = to_tushare_date(trade_date)
+        for limit_type in LIMIT_TYPES:
+            current_call += 1
+            print(f"  [{current_call}/{total_calls}] {tushare_date} limit_type={limit_type}")
+            rate_limiter.wait()
+            try:
+                df = pro.limit_list_d(
+                    trade_date=tushare_date,
+                    limit_type=limit_type,
+                    fields=FIELDS,
+                )
+            except Exception as exc:
+                print(f"⚠️ 获取失败: {tushare_date} {limit_type}, 错误: {exc}")
+                continue
+
+            if df is None or df.empty:
+                continue
+            frames.append(normalize_data(df))
+
+    if not frames:
+        return pd.DataFrame()
+    return pd.concat(frames, ignore_index=True).drop_duplicates(
+        subset=["trade_date", "ts_code", "limit"],
+        keep="first",
+    )
+
+
+def save_data(engine, df, trade_dates):
+    if not trade_dates:
+        return 0
+
+    with engine.begin() as conn:
+        delete_params = {f"date_{idx}": value for idx, value in enumerate(trade_dates)}
+        placeholders = ", ".join(f":date_{idx}" for idx in range(len(trade_dates)))
+        delete_sql = text(f"DELETE FROM `{TABLE_NAME}` WHERE trade_date IN ({placeholders})")
+        delete_result = conn.execute(delete_sql, delete_params)
+        print(f"🧹 已清理旧数据 {delete_result.rowcount} 条。")
+
+        if df.empty:
+            print("ℹ️ 本次没有可写入的 limit_list_d 数据。")
+            return 0
+
+        df.to_sql(
+            TABLE_NAME,
+            conn,
+            if_exists="append",
+            index=False,
+            chunksize=1000,
+            method="multi",
+        )
+        return len(df)
+
+
+def run(start_date, end_date):
+    start = parse_date(start_date)
+    end = parse_date(end_date)
+    if start > end:
+        raise ValueError("开始日期不能大于结束日期")
+
+    if end < TUSHARE_START_DATE:
+        print(f"ℹ️ limit_list_d 接口从 {TUSHARE_START_DATE} 开始处理，当前区间无需同步。")
+        return True
+    if start < TUSHARE_START_DATE:
+        print(f"ℹ️ start_date 早于 {TUSHARE_START_DATE}，已自动调整。")
+        start = TUSHARE_START_DATE
+
+    engine = get_engine()
+    init_db(engine)
+
+    trade_dates = get_trade_dates(engine, start, end)
+    print(f"📅 需要处理 {len(trade_dates)} 个交易日/日期：{start} ~ {end}")
+    if not trade_dates:
+        return True
+
+    df = fetch_limit_list_d(trade_dates)
+    inserted = save_data(engine, df, trade_dates)
+    print(f"✅ {TABLE_NAME} 同步完成，写入 {inserted} 条。")
+    return True
+
+
+if __name__ == "__main__":
+    START_DATE = "20260513"  # 开始日期，格式 YYYYMMDD 或 YYYY-MM-DD
+    END_DATE = "20260515"  # 结束日期，格式 YYYYMMDD 或 YYYY-MM-DD
+
+    run(START_DATE, END_DATE)

limit_up.py

@@ -1,98 +1,717 @@
 import pandas as pd
-import numpy as np
+from sqlalchemy import text
-from sqlalchemy import create_engine, text
+import argparse
+import tushare as ts
+from datetime import date
+from collections import deque
+import time
+
+from stock_config import TUSHARE_TOKEN, get_engine
+
+# --- 数据库配置 ---
+# 请根据实际情况修改数据库连接信息
+# DB_CONN_STR = 'mysql+pymysql://root:password@127.0.0.1:3306/stock_data'
+LOCAL_START_DATE = date(2018, 1, 1)
+TUSHARE_LIMIT_LIST_START_DATE = date(2020, 1, 1)
+TUSHARE_LIMIT_LIST_MAX_CALLS_PER_MINUTE = 200
+COMBINED_TABLE = "stock_limit_combined_data"
 
 
-def init_db(engine):
+class MinuteRateLimiter:
+    def __init__(self, max_calls):
+        self.max_calls = max_calls
+        self.call_times = deque()
+
+    def wait(self):
+        now = time.monotonic()
+        while self.call_times and now - self.call_times[0] >= 60:
+            self.call_times.popleft()
+        if len(self.call_times) >= self.max_calls:
+            sleep_seconds = 60 - (now - self.call_times[0]) + 0.1
+            print(f"⏳ Tushare 限流等待 {sleep_seconds:.1f} 秒，避免超过每分钟 {self.max_calls} 次。")
+            time.sleep(sleep_seconds)
+            now = time.monotonic()
+            while self.call_times and now - self.call_times[0] >= 60:
+                self.call_times.popleft()
+        self.call_times.append(time.monotonic())
+
+
+def normalize_trade_dates(trade_dates):
+    return sorted({pd.to_datetime(trade_date).date() for trade_date in trade_dates})
+
+
+def filter_dates_by_start(trade_dates, start_date, source_name):
+    normalized_dates = normalize_trade_dates(trade_dates)
+    filtered_dates = [trade_date for trade_date in normalized_dates if trade_date >= start_date]
+    skipped_count = len(normalized_dates) - len(filtered_dates)
+    if skipped_count:
+        print(f"ℹ️ {source_name} 跳过 {skipped_count} 个早于 {start_date} 的交易日。")
+    return filtered_dates
+
+
+def init_db(engine, full_maintenance=False):
     """
     初始化数据库表结构
     """
-    create_table_sql = """
+    with engine.begin() as conn:
-    CREATE TABLE IF NOT EXISTS `stock_limit_data` (
+        print(f"  - 确保 {COMBINED_TABLE} 存在 ...")
+        conn.execute(text(get_combined_table_sql()))
+        optimize_combined_table(conn, full_maintenance=full_maintenance)
+
+
+def get_combined_table_sql():
+    return f"""
+    CREATE TABLE IF NOT EXISTS `{COMBINED_TABLE}` (
         `id` BIGINT AUTO_INCREMENT PRIMARY KEY,
         `trade_date` DATE NOT NULL COMMENT '交易日期',
-        `symbol` VARCHAR(10) NOT NULL COMMENT '股票代码',
+        `symbol` VARCHAR(10) NOT NULL COMMENT '股票代码，以stock_daily为准',
-        `ts_code` VARCHAR(20) NOT NULL COMMENT 'TS股票代码',
+        `ts_code` VARCHAR(20) NOT NULL COMMENT 'TS股票代码，以stock_daily为准',
-        `name` VARCHAR(50) NOT NULL COMMENT '股票名称',
+        `name` VARCHAR(50) NOT NULL COMMENT '股票名称，以stock_daily关联基础表为准',
-        `status` TINYINT NOT NULL COMMENT '状态：1-涨停，2-炸板',
+        `final_status` TINYINT NOT NULL COMMENT '融合状态：1-涨停，2-炸板',
-        `limit_price` DECIMAL(16, 4) COMMENT '计算出的理论涨停价',
+        `source_count` INT NOT NULL DEFAULT 0 COMMENT '命中的数据源数量',
-        `close` DECIMAL(16, 4) COMMENT '实际收盘价',
+        `source_flags` VARCHAR(50) COMMENT '命中的数据源：calc,tushare',
+        `calc_status` TINYINT COMMENT '本地行情计算状态：1-涨停，2-炸板',
+        `calc_limit_price` DECIMAL(16, 4) COMMENT '本地行情计算涨停价',
+        `calc_high` DECIMAL(16, 4) COMMENT 'stock_daily最高价',
+        `calc_close` DECIMAL(16, 4) COMMENT 'stock_daily收盘价',
+        `calc_pre_close` DECIMAL(16, 4) COMMENT 'stock_daily前收盘价',
+        `calc_vol` BIGINT COMMENT 'stock_daily成交量',
         `market` VARCHAR(20) COMMENT '所属板块',
-        UNIQUE KEY `uniq_date_code` (`trade_date`, `ts_code`)
+        `tushare_limit_type` VARCHAR(1) COMMENT 'Tushare-D跌停U涨停Z炸板',
-    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
+        `tushare_industry` VARCHAR(100) COMMENT 'Tushare-所属行业',
+        `tushare_name` VARCHAR(50) COMMENT 'Tushare-名称',
+        `tushare_close` DECIMAL(16, 4) COMMENT 'Tushare-收盘价',
+        `tushare_pct_chg` DECIMAL(10, 4) COMMENT 'Tushare-涨跌幅',
+        `tushare_amount` DECIMAL(24, 4) COMMENT 'Tushare-成交额',
+        `tushare_limit_amount` DECIMAL(24, 4) COMMENT 'Tushare-板上成交金额',
+        `tushare_float_mv` DECIMAL(24, 4) COMMENT 'Tushare-流通市值',
+        `tushare_total_mv` DECIMAL(24, 4) COMMENT 'Tushare-总市值',
+        `tushare_turnover_ratio` DECIMAL(10, 4) COMMENT 'Tushare-换手率',
+        `tushare_fd_amount` DECIMAL(24, 4) COMMENT 'Tushare-封单金额',
+        `tushare_first_time` VARCHAR(20) COMMENT 'Tushare-首次封板时间',
+        `tushare_last_time` VARCHAR(20) COMMENT 'Tushare-最后封板时间',
+        `tushare_open_times` INT COMMENT 'Tushare-炸板/开板次数',
+        `tushare_up_stat` VARCHAR(50) COMMENT 'Tushare-涨停统计',
+        `tushare_limit_times` INT COMMENT 'Tushare-连板数',
+        UNIQUE KEY `uniq_date_code` (`trade_date`, `ts_code`),
+        KEY `idx_trade_date` (`trade_date`),
+        KEY `idx_symbol` (`symbol`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci;
     """
-    with engine.begin() as conn:
+
-        conn.execute(text(create_table_sql))
+
+def get_table_columns(conn, table_name):
+    return {
+        row[0]
+        for row in conn.execute(text(f"SHOW COLUMNS FROM `{table_name}`")).fetchall()
+    }
+
+
+def get_column_collation(conn, table_name, column_name):
+    row = conn.execute(
+        text(
+            """
+            SELECT collation_name
+            FROM information_schema.columns
+            WHERE table_schema = DATABASE()
+              AND table_name = :table_name
+              AND column_name = :column_name
+            """
+        ),
+        {"table_name": table_name, "column_name": column_name},
+    ).first()
+    return row[0] if row else None
+
+
+def index_exists(conn, table_name, index_name):
+    return conn.execute(
+        text(
+            """
+            SELECT 1
+            FROM information_schema.statistics
+            WHERE table_schema = DATABASE()
+              AND table_name = :table_name
+              AND index_name = :index_name
+            LIMIT 1
+            """
+        ),
+        {"table_name": table_name, "index_name": index_name},
+    ).first() is not None
+
+
+def get_index_definition(conn, table_name, index_name):
+    rows = conn.execute(
+        text(
+            """
+            SELECT column_name, non_unique
+            FROM information_schema.statistics
+            WHERE table_schema = DATABASE()
+              AND table_name = :table_name
+              AND index_name = :index_name
+            ORDER BY seq_in_index
+            """
+        ),
+        {"table_name": table_name, "index_name": index_name},
+    ).fetchall()
+    if not rows:
+        return None
+    return {
+        "columns": [row[0] for row in rows],
+        "unique": rows[0][1] == 0,
+    }
+
+
+def index_matches(conn, table_name, index_name, columns, unique=False):
+    index_def = get_index_definition(conn, table_name, index_name)
+    if not index_def:
+        return False
+    return index_def["columns"] == columns and (not unique or index_def["unique"])
+
+
+def ensure_combined_columns(conn):
+    existing_columns = get_table_columns(conn, COMBINED_TABLE)
+    columns = {
+        "calc_vol": "BIGINT COMMENT 'stock_daily成交量'",
+    }
+
+    for column_name, column_def in columns.items():
+        if column_name not in existing_columns:
+            conn.execute(
+                text(
+                    f"ALTER TABLE `{COMBINED_TABLE}` "
+                    f"ADD COLUMN `{column_name}` {column_def} AFTER `calc_pre_close`"
+                )
+            )
+
+    ensure_combined_join_collation(conn)
+
+
+def ensure_combined_join_collation(conn):
+    daily_collation = get_column_collation(conn, "stock_daily", "ts_code")
+    combined_collation = get_column_collation(conn, COMBINED_TABLE, "ts_code")
+    if not daily_collation or not combined_collation or daily_collation == combined_collation:
+        return
+
+    print(f"  - 对齐 ts_code collation：{combined_collation} -> {daily_collation} ...")
+    conn.execute(
+        text(
+            f"""
+            ALTER TABLE `{COMBINED_TABLE}`
+            MODIFY `ts_code` VARCHAR(20)
+            CHARACTER SET utf8mb4 COLLATE {daily_collation}
+            NOT NULL COMMENT 'TS股票代码，以stock_daily为准'
+            """
+        )
+    )
+
+
+def sync_combined_vol_from_daily(conn):
+    conn.execute(
+        text(
+            f"""
+            UPDATE `{COMBINED_TABLE}` c
+            JOIN `stock_daily` d
+              ON c.ts_code = d.ts_code
+             AND c.trade_date = d.trade_date
+            SET c.symbol = d.symbol,
+                c.calc_vol = d.vol
+            """
+        )
+    )
+
+
+def dedupe_combined_table(conn):
+    conn.execute(
+        text(
+            f"""
+            DELETE c
+            FROM `{COMBINED_TABLE}` c
+            LEFT JOIN `stock_daily` d
+              ON c.ts_code = d.ts_code
+             AND c.trade_date = d.trade_date
+            WHERE d.ts_code IS NULL
+            """
+        )
+    )
+    conn.execute(
+        text(
+            f"""
+            DELETE c1
+            FROM `{COMBINED_TABLE}` c1
+            JOIN `{COMBINED_TABLE}` c2
+              ON c1.trade_date = c2.trade_date
+             AND c1.ts_code = c2.ts_code
+            WHERE
+                (
+                    COALESCE(c1.source_count, 0) * 10
+                    + IF(c1.calc_status IS NOT NULL, 4, 0)
+                    + IF(c1.tushare_limit_type IS NOT NULL, 2, 0)
+                    + IF(c1.calc_vol IS NOT NULL, 1, 0)
+                ) <
+                (
+                    COALESCE(c2.source_count, 0) * 10
+                    + IF(c2.calc_status IS NOT NULL, 4, 0)
+                    + IF(c2.tushare_limit_type IS NOT NULL, 2, 0)
+                    + IF(c2.calc_vol IS NOT NULL, 1, 0)
+                )
+                OR (
+                    (
+                        COALESCE(c1.source_count, 0) * 10
+                        + IF(c1.calc_status IS NOT NULL, 4, 0)
+                        + IF(c1.tushare_limit_type IS NOT NULL, 2, 0)
+                        + IF(c1.calc_vol IS NOT NULL, 1, 0)
+                    ) =
+                    (
+                        COALESCE(c2.source_count, 0) * 10
+                        + IF(c2.calc_status IS NOT NULL, 4, 0)
+                        + IF(c2.tushare_limit_type IS NOT NULL, 2, 0)
+                        + IF(c2.calc_vol IS NOT NULL, 1, 0)
+                    )
+                    AND c1.id < c2.id
+                )
+            """
+        )
+    )
+
+
+def ensure_combined_indexes(conn, full_maintenance=False):
+    needs_unique_rebuild = index_exists(
+        conn, COMBINED_TABLE, "uniq_date_code"
+    ) and not index_matches(
+        conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True
+    )
+    needs_unique_create = not index_matches(
+        conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True
+    )
+
+    if full_maintenance or needs_unique_rebuild or needs_unique_create:
+        print("  - 清理历史重复/无效融合数据 ...")
+        dedupe_combined_table(conn)
+
+    if index_exists(conn, COMBINED_TABLE, "uniq_date_code") and not index_matches(
+        conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True
+    ):
+        print("  - 重建 uniq_date_code 唯一索引 ...")
+        conn.execute(text(f"ALTER TABLE `{COMBINED_TABLE}` DROP INDEX `uniq_date_code`"))
+
+    if not index_matches(
+        conn, COMBINED_TABLE, "uniq_date_code", ["trade_date", "ts_code"], unique=True
+    ):
+        print("  - 创建 uniq_date_code 唯一索引 ...")
+        conn.execute(
+            text(
+                f"ALTER TABLE `{COMBINED_TABLE}` "
+                "ADD UNIQUE KEY `uniq_date_code` (`trade_date`, `ts_code`)"
+            )
+        )
+    if not index_exists(conn, COMBINED_TABLE, "idx_trade_date"):
+        print("  - 创建 idx_trade_date 索引 ...")
+        conn.execute(
+            text(f"ALTER TABLE `{COMBINED_TABLE}` ADD KEY `idx_trade_date` (`trade_date`)")
+        )
+    if not index_exists(conn, COMBINED_TABLE, "idx_symbol"):
+        print("  - 创建 idx_symbol 索引 ...")
+        conn.execute(
+            text(f"ALTER TABLE `{COMBINED_TABLE}` ADD KEY `idx_symbol` (`symbol`)")
+        )
+    if not index_exists(conn, COMBINED_TABLE, "idx_trade_date_status"):
+        print("  - 创建 idx_trade_date_status 索引 ...")
+        conn.execute(
+            text(
+                f"ALTER TABLE `{COMBINED_TABLE}` "
+                "ADD KEY `idx_trade_date_status` (`trade_date`, `final_status`)"
+            )
+        )
+
+
+def optimize_combined_table(conn, full_maintenance=False):
+    print("  - 检查字段 ...")
+    ensure_combined_columns(conn)
+    if full_maintenance:
+        print("  - 从 stock_daily 全量回填 calc_vol ...")
+        sync_combined_vol_from_daily(conn)
+    else:
+        print("  - 跳过全量 calc_vol 回填，当前批次写入时会直接带入 vol。")
+    print("  - 检查索引 ...")
+    ensure_combined_indexes(conn, full_maintenance=full_maintenance)
+    print("  - 校验重复键 ...")
+    validate_no_duplicate_combined_rows(conn)
+
+
+def validate_no_duplicate_combined_rows(conn):
+    duplicate_count = conn.execute(
+        text(
+            f"""
+            SELECT COUNT(*)
+            FROM (
+                SELECT trade_date, ts_code
+                FROM `{COMBINED_TABLE}`
+                GROUP BY trade_date, ts_code
+                HAVING COUNT(*) > 1
+            ) duplicated
+            """
+        )
+    ).scalar()
+    if duplicate_count:
+        raise RuntimeError(f"{COMBINED_TABLE} 存在 {duplicate_count} 组重复 trade_date+ts_code 数据")
+
+
+def validate_no_duplicate_frame_rows(df, frame_name):
+    if df.empty:
+        return
+    missing_keys = df["trade_date"].isna() | df["ts_code"].isna()
+    if missing_keys.any():
+        examples = df.loc[missing_keys, ["trade_date", "ts_code"]].head(5).to_dict("records")
+        raise ValueError(f"{frame_name} 存在空 trade_date 或 ts_code，示例: {examples}")
+    duplicated = df.duplicated(subset=["trade_date", "ts_code"], keep=False)
+    if duplicated.any():
+        examples = df.loc[duplicated, ["trade_date", "ts_code"]].head(5).to_dict("records")
+        raise ValueError(f"{frame_name} 存在重复 trade_date+ts_code，示例: {examples}")
+
+
+def fetch_tushare_limit_list(trade_dates):
+    tushare_trade_dates = filter_dates_by_start(
+        trade_dates,
+        TUSHARE_LIMIT_LIST_START_DATE,
+        "Tushare limit_list_d",
+    )
+    if not tushare_trade_dates:
+        print("ℹ️ 无需调用 Tushare limit_list_d。")
+        return pd.DataFrame(columns=["trade_date", "symbol", "ts_code"])
+
+    ts.set_token(TUSHARE_TOKEN)
+    pro = ts.pro_api()
+
+    frames = []
+    rate_limiter = MinuteRateLimiter(TUSHARE_LIMIT_LIST_MAX_CALLS_PER_MINUTE)
+    total_calls = len(tushare_trade_dates) * 2
+    current_call = 0
+    fields = (
+        "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,"
+        "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times,limit"
+    )
+    print(
+        f"🔎 开始获取 Tushare limit_list_d：{len(tushare_trade_dates)} 个交易日，"
+        f"预计 {total_calls} 次调用，限速 {TUSHARE_LIMIT_LIST_MAX_CALLS_PER_MINUTE} 次/分钟。"
+    )
+    for trade_date in tushare_trade_dates:
+        tushare_date = pd.to_datetime(trade_date).strftime("%Y%m%d")
+        for limit_type in ("U", "Z"):
+            current_call += 1
+            print(f"  [{current_call}/{total_calls}] 获取 {tushare_date} limit_type={limit_type} ...")
+            rate_limiter.wait()
+            try:
+                ts_df = pro.limit_list_d(
+                    trade_date=tushare_date,
+                    limit_type=limit_type,
+                    fields=fields,
+                )
+            except Exception as exc:
+                print(f"⚠️ Tushare limit_list_d 获取失败: {tushare_date} {limit_type}, 错误: {exc}")
+                continue
+
+            if ts_df is None or ts_df.empty:
+                print(f"  [{current_call}/{total_calls}] {tushare_date} {limit_type} 无数据。")
+                continue
+
+            normalized_df = normalize_tushare_limit_list(ts_df)
+            if not normalized_df.empty:
+                print(f"  [{current_call}/{total_calls}] {tushare_date} {limit_type} 获取 {len(normalized_df)} 条。")
+                frames.append(normalized_df)
+
+    if not frames:
+        print("ℹ️ Tushare limit_list_d 未返回有效数据。")
+        return pd.DataFrame(columns=["trade_date", "symbol", "ts_code"])
+
+    result_df = pd.concat(frames, ignore_index=True).drop_duplicates(
+        subset=["trade_date", "ts_code", "tushare_limit_type"],
+        keep="first",
+    )
+    print(f"✅ Tushare limit_list_d 获取完成：{len(result_df)} 条有效记录。")
+    return result_df
+
+
+def normalize_tushare_limit_list(ts_df):
+    column_map = {
+        "limit": "tushare_limit_type",
+        "industry": "tushare_industry",
+        "name": "tushare_name",
+        "close": "tushare_close",
+        "pct_chg": "tushare_pct_chg",
+        "amount": "tushare_amount",
+        "limit_amount": "tushare_limit_amount",
+        "float_mv": "tushare_float_mv",
+        "total_mv": "tushare_total_mv",
+        "turnover_ratio": "tushare_turnover_ratio",
+        "fd_amount": "tushare_fd_amount",
+        "first_time": "tushare_first_time",
+        "last_time": "tushare_last_time",
+        "open_times": "tushare_open_times",
+        "up_stat": "tushare_up_stat",
+        "limit_times": "tushare_limit_times",
+    }
+
+    required_columns = {"trade_date", "ts_code", "limit"}
+    if not required_columns.issubset(ts_df.columns):
+        missing_columns = sorted(required_columns - set(ts_df.columns))
+        print(f"⚠️ Tushare 返回字段缺失，已跳过: {', '.join(missing_columns)}")
+        return pd.DataFrame(columns=["trade_date", "symbol", "ts_code"])
+
+    available_map = {
+        source: target for source, target in column_map.items() if source in ts_df.columns
+    }
+    normalized_df = ts_df[["trade_date", "ts_code", *available_map]].rename(
+        columns=available_map
+    ).copy()
+    normalized_df["trade_date"] = pd.to_datetime(
+        normalized_df["trade_date"], format="%Y%m%d"
+    ).dt.date
+    normalized_df["ts_code"] = normalized_df["ts_code"].astype(str).str.strip()
+    normalized_df["symbol"] = normalized_df["ts_code"].str.split(".").str[0].str.zfill(6)
+
+    for column in [
+        "tushare_close",
+        "tushare_pct_chg",
+        "tushare_amount",
+        "tushare_limit_amount",
+        "tushare_float_mv",
+        "tushare_total_mv",
+        "tushare_turnover_ratio",
+        "tushare_fd_amount",
+        "tushare_open_times",
+        "tushare_limit_times",
+    ]:
+        if column in normalized_df.columns:
+            normalized_df[column] = pd.to_numeric(normalized_df[column], errors="coerce")
+
+    for column in [
+        "tushare_limit_type",
+        "tushare_industry",
+        "tushare_name",
+        "tushare_first_time",
+        "tushare_last_time",
+        "tushare_up_stat",
+    ]:
+        if column in normalized_df.columns:
+            normalized_df[column] = normalized_df[column].astype(str)
+
+    return normalized_df
+
+
+def build_combined_limit_data(daily_df, calc_df, tushare_df):
+    tushare_columns = [
+        "tushare_limit_type",
+        "tushare_industry",
+        "tushare_name",
+        "tushare_close",
+        "tushare_pct_chg",
+        "tushare_amount",
+        "tushare_limit_amount",
+        "tushare_float_mv",
+        "tushare_total_mv",
+        "tushare_turnover_ratio",
+        "tushare_fd_amount",
+        "tushare_first_time",
+        "tushare_last_time",
+        "tushare_open_times",
+        "tushare_up_stat",
+        "tushare_limit_times",
+    ]
+
+    for column in tushare_columns:
+        if column not in tushare_df.columns:
+            tushare_df[column] = None
+
+    calc_key_df = pd.DataFrame(columns=["trade_date", "symbol"])
+    if not calc_df.empty:
+        calc_key_df = calc_df[["trade_date", "symbol"]].copy()
+    tushare_key_df = pd.DataFrame(columns=["trade_date", "symbol"])
+    if not tushare_df.empty:
+        tushare_key_df = tushare_df[["trade_date", "symbol"]].copy()
+
+    keys_df = pd.concat(
+        [calc_key_df, tushare_key_df],
+        ignore_index=True,
+    ).drop_duplicates()
+    if keys_df.empty:
+        return pd.DataFrame()
+
+    for frame in [daily_df, calc_df, tushare_df, keys_df]:
+        if "trade_date" in frame.columns:
+            frame["trade_date"] = pd.to_datetime(frame["trade_date"]).dt.date
+        if "symbol" in frame.columns:
+            frame["symbol"] = frame["symbol"].astype(str).str.zfill(6)
+    keys_df = keys_df.drop_duplicates(subset=["trade_date", "symbol"])
+
+    canonical_df = daily_df[[
+        "trade_date",
+        "symbol",
+        "ts_code",
+        "name",
+        "high",
+        "close",
+        "pre_close",
+        "vol",
+        "market",
+    ]].drop_duplicates(subset=["trade_date", "symbol"])
+    combined_df = keys_df.merge(
+        canonical_df,
+        how="inner",
+        on=["trade_date", "symbol"],
+    )
+    if combined_df.empty:
+        return pd.DataFrame()
+
+    if not calc_df.empty:
+        calc_merge_df = calc_df[[
+            "trade_date",
+            "symbol",
+            "status",
+            "limit_price",
+        ]].rename(columns={
+            "status": "calc_status",
+            "limit_price": "calc_limit_price",
+        })
+        combined_df = combined_df.merge(
+            calc_merge_df,
+            how="left",
+            on=["trade_date", "symbol"],
+        )
+    else:
+        combined_df["calc_status"] = None
+        combined_df["calc_limit_price"] = None
+
+    if not tushare_df.empty:
+        tushare_merge_df = tushare_df[
+            ["trade_date", "symbol", *tushare_columns]
+        ].sort_values("tushare_limit_type").drop_duplicates(
+            subset=["trade_date", "symbol"],
+            keep="last",
+        )
+        combined_df = combined_df.merge(
+            tushare_merge_df,
+            how="left",
+            on=["trade_date", "symbol"],
+        )
+    else:
+        for column in tushare_columns:
+            combined_df[column] = None
+
+    combined_df["calc_high"] = combined_df["high"]
+    combined_df["calc_close"] = combined_df["close"]
+    combined_df["calc_pre_close"] = combined_df["pre_close"]
+    combined_df["calc_vol"] = combined_df["vol"]
+
+    combined_df["final_status"] = combined_df.apply(resolve_final_status, axis=1)
+    combined_df["source_flags"] = combined_df.apply(resolve_source_flags, axis=1)
+    combined_df["source_count"] = combined_df["source_flags"].apply(
+        lambda value: 0 if not value else len(value.split(","))
+    )
+    combined_df["name"] = combined_df["name"].astype(str).str.strip()
+
+    result_df = combined_df[[
+        "trade_date",
+        "symbol",
+        "ts_code",
+        "name",
+        "final_status",
+        "source_count",
+        "source_flags",
+        "calc_status",
+        "calc_limit_price",
+        "calc_high",
+        "calc_close",
+        "calc_pre_close",
+        "calc_vol",
+        "market",
+        *tushare_columns,
+    ]].copy()
+    result_df = result_df.sort_values(
+        ["trade_date", "ts_code", "source_count"],
+        ascending=[True, True, False],
+    )
+    return result_df.drop_duplicates(subset=["trade_date", "ts_code"], keep="first")
+
+
+def resolve_final_status(row):
+    if pd.notna(row.get("calc_status")):
+        return int(row["calc_status"])
+    if row.get("tushare_limit_type") == "Z":
+        return 2
+    return 1
+
+
+def resolve_source_flags(row):
+    flags = []
+    if pd.notna(row.get("calc_status")):
+        flags.append("calc")
+    if pd.notna(row.get("tushare_limit_type")):
+        flags.append("tushare")
+    return ",".join(flags)
 
 
 def analyze_and_save_limits(start_date, end_date):
     """
     计算并在数据库中落地涨停、炸板详情
     """
-    # 数据库连接配置
+    engine = get_engine()
-    engine = create_engine(
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
 
+    # 0. 初始化检查表结构
+    print(f"🚀 开始处理涨停/炸板数据：({start_date} -> {end_date}]")
+    print(f"🧱 初始化并检查融合表 {COMBINED_TABLE} ...")
     init_db(engine)
+    print("✅ 表结构、历史重复数据和唯一索引检查完成。")
 
-    # 1. 获取行情数据
+    requested_start_date = pd.to_datetime(start_date).date()
-    query = f"""
+    if requested_start_date < LOCAL_START_DATE:
+        print(f"ℹ️ 本地 stock_daily 暂按 {LOCAL_START_DATE} 起处理，已忽略更早日期。")
+
+    # 1. 获取基础数据和行情数据
+    query = text("""
     SELECT 
         d.trade_date, d.symbol, d.ts_code, b.name, 
-        d.high, d.close, d.pre_close, b.market
+        d.high, d.close, d.pre_close, d.vol, b.market
     FROM stock_daily d
-    JOIN stock_basic b ON d.ts_code = b.ts_code
+    JOIN stock_basic b
-    WHERE d.trade_date >= '{start_date}' 
+      ON d.ts_code COLLATE utf8mb4_unicode_ci = b.ts_code COLLATE utf8mb4_unicode_ci
-      AND d.trade_date <= '{end_date}'
+    WHERE d.trade_date > :start_date
+      AND d.trade_date <= :end_date
+      AND d.trade_date >= :local_start_date
       AND d.pre_close IS NOT NULL
     ORDER BY d.trade_date ASC
-    """
+    """)
 
     try:
-        df = pd.read_sql(query, engine)
+        print("📥 读取 stock_daily 与 stock_basic 数据 ...")
+        df = pd.read_sql(
+            query,
+            engine,
+            params={
+                "start_date": start_date,
+                "end_date": end_date,
+                "local_start_date": LOCAL_START_DATE,
+            },
+        )
         if df.empty:
-            print(f"⚠️ 区间 [{start_date} , {end_date}] 内未获取到数据。")
+            print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。")
-            return
+            return True
+        target_dates = df['trade_date'].astype(str).unique()
+        print(f"✅ 读取完成：{len(df)} 行行情，覆盖 {len(target_dates)} 个交易日。")
 
-        # 强制类型转换
+        # 2. 涨停价计算规则
-        df['pre_close'] = pd.to_numeric(df['pre_close'], errors='coerce')
-        df['high'] = pd.to_numeric(df['high'], errors='coerce')
-        df['close'] = pd.to_numeric(df['close'], errors='coerce')
-
-        # 2. 涨停价计算规则 (已取消 ST 判定)
         def calc_limit(row):
-            m = str(row['market'])
+            m, p = str(row['market']), float(row['pre_close'])
-            p = row['pre_close']
+            # 主板10%，创业板/科创板20%，北交所30%
-
+            ratio = 1.2 if ('创业板' in m or '科创板' in m) else (1.3 if '北交所' in m else 1.1)
-            if pd.isna(p) or p <= 0: return 0.0
-
-            # 仅保留板块判定
-            if '创业板' in m or '科创板' in m:
-                ratio = 1.2
-            elif '北交所' in m:
-                ratio = 1.3
-            else:
-                ratio = 1.1  # 普通主板（含ST股现在也按10%计算）
-
-            # 采用标准四舍五入
             return round(p * ratio, 2)
 
         df['limit_price'] = df.apply(calc_limit, axis=1)
 
-        # 3. 筛选逻辑：利用 abs 差值判断，解决浮点数精度问题
+        # 3. 筛选涨停(1)和炸板(2)
-        # 只要最高价触及理论涨停价（误差0.01以内）即视为触板
+        df['is_limit'] = (df['high'] == df['limit_price']) & (df['close'] == df['limit_price'])
-        is_at_limit = (df['high'] - df['limit_price']).abs() < 0.011
+        df['is_broken'] = (df['high'] == df['limit_price']) & (df['close'] < df['limit_price'])
-        is_close_at_limit = (df['close'] - df['limit_price']).abs() < 0.011
 
-        df['is_limit'] = is_at_limit & is_close_at_limit
-        df['is_broken'] = is_at_limit & (~is_close_at_limit)
-
-        # 分离结果
         limit_df = df[df['is_limit']].copy()
         limit_df['status'] = 1
 
@@ -100,35 +719,90 @@ def analyze_and_save_limits(start_date, end_date):
         broken_df['status'] = 2
 
         save_df = pd.concat([limit_df, broken_df])
+        print(
+            f"🧮 本地计算完成：涨停 {len(limit_df)} 条，炸板 {len(broken_df)} 条，"
+            f"合计 {len(save_df)} 条。"
+        )
+
+        tushare_df = fetch_tushare_limit_list(target_dates)
+        print("🔗 融合本地计算结果与 Tushare 数据 ...")
+        combined_df = build_combined_limit_data(df.copy(), save_df.copy(), tushare_df.copy())
+        validate_no_duplicate_frame_rows(combined_df, "融合待入库数据")
+        print(f"✅ 融合完成：待入库 {len(combined_df)} 条，重复键检查通过。")
+
+        with engine.begin() as conn:
+            delete_params = {
+                f"date_{idx}": value for idx, value in enumerate(target_dates)
+            }
+            placeholders = ", ".join(f":date_{idx}" for idx in range(len(target_dates)))
+            combined_delete_query = text(
+                f"DELETE FROM {COMBINED_TABLE} WHERE trade_date IN ({placeholders})"
+            )
+            print(f"🧹 清理 {len(target_dates)} 个交易日的旧融合数据 ...")
+            delete_result = conn.execute(combined_delete_query, delete_params)
+            print(f"✅ 旧数据清理完成：删除 {delete_result.rowcount} 条。")
+
+            if not combined_df.empty:
+                print(f"💾 写入 {COMBINED_TABLE}：{len(combined_df)} 条 ...")
+                combined_df.to_sql(
+                    COMBINED_TABLE,
+                    conn,
+                    if_exists='append',
+                    index=False,
+                    chunksize=1000,
+                    method="multi",
+                )
+            else:
+                print("ℹ️ 融合结果为空，本次不写入新数据。")
+            validate_no_duplicate_combined_rows(conn)
+            print("✅ 写入后数据库重复键校验通过。")
 
         if not save_df.empty:
-            final_df = save_df[[
-                'trade_date', 'symbol', 'ts_code', 'name',
-                'status', 'limit_price', 'close', 'market'
-            ]]
-
-            # 4. 事务入库：先删后插
-            with engine.begin() as conn:
-                target_dates = final_df['trade_date'].astype(str).unique()
-                date_list = ", ".join([f"'{d}'" for d in target_dates])
-
-                delete_query = f"DELETE FROM stock_limit_data WHERE trade_date IN ({date_list})"
-                conn.execute(text(delete_query))
-
-                final_df.to_sql('stock_limit_data', conn, if_exists='append', index=False)
-
             print(f"✅ 处理成功！涉及交易日: {len(target_dates)} 天")
-            print(f"📊 记录详情: 总计 {len(final_df)} 条 (涨停: {len(limit_df)} / 炸板: {len(broken_df)})")
+            print(f"📊 融合表 {COMBINED_TABLE} 入库: {len(combined_df)} 条记录。")
         else:
-            print("ℹ️ 该时段内全市场无涨停或炸板记录。")
+            print(f"ℹ️ 该时段内本地行情无涨停或炸板记录，已清理 {len(target_dates)} 个交易日的融合表旧结果。")
+            print(f"📊 融合表 {COMBINED_TABLE} 入库: {len(combined_df)} 条记录。")
+        return True
 
     except Exception as e:
         print(f"❌ 运行失败: {e}")
+        return False
 
 
+def cleanup_combined_data():
+    engine = get_engine()
+    try:
+        init_db(engine, full_maintenance=True)
+        print(f"✅ {COMBINED_TABLE} 已完成历史重复数据清理、唯一索引检查和 calc_vol 回填。")
+        return True
+    except Exception as e:
+        print(f"❌ 清理失败: {e}")
+        return False
+
+
+# ============================================================
+# 用户配置区
+# ============================================================
 if __name__ == "__main__":
-    # 配置日期区间
+    parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据")
-    START_DATE = '2024-09-23'
+    parser.add_argument("--start-date", default="2026-05-13", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
-    END_DATE = '2026-04-27'
+    parser.add_argument("--end-date", default="2026-05-15", help="结束日期，格式 YYYY-MM-DD")
+    parser.add_argument(
+        "--cleanup-only",
+        action="store_true",
+        help="仅清理 stock_limit_combined_data 历史重复数据并从 stock_daily 回填 calc_vol",
+    )
+    args = parser.parse_args()
+
+    if args.cleanup_only:
+        cleanup_combined_data()
+        raise SystemExit(0)
+
+    # 开始日期：计算从该日期的下一个交易日开始
+    START_DATE = args.start_date
+
+    # 结束日期
+    END_DATE = args.end_date
 
     analyze_and_save_limits(START_DATE, END_DATE)

luoji.py

@@ -1,106 +0,0 @@
-import pandas as pd
-from sqlalchemy import create_engine
-
-
-def analyze_stock_performance(start_date, end_date):
-    # --- 数据库配置 ---
-    try:
-        engine = create_engine(
-            "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-            connect_args={
-                "password": "aA%8023321088"
-            },
-            pool_pre_ping=True
-        )
-
-        # 1. 获取涨停当日股票及价格
-        limit_query = f"""
-        SELECT l.ts_code, l.name, d.open AS limit_day_open, d.high AS limit_day_high
-        FROM stock_limit_data l
-        LEFT JOIN stock_daily d ON l.ts_code = d.ts_code AND l.trade_date = d.trade_date
-        WHERE l.trade_date = '{start_date}' AND l.status = 1
-        """
-        df_limit = pd.read_sql(limit_query, engine)
-
-        if df_limit.empty:
-            print(f"提示：{start_date} 当日没有涨停记录。")
-            return
-
-        # 2. 获取行情区间数据
-        codes = df_limit['ts_code'].unique().tolist()
-        codes_tuple = f"('{codes[0]}')" if len(codes) == 1 else str(tuple(codes))
-        daily_query = f"""
-        SELECT ts_code, trade_date, high, low, close
-        FROM stock_daily
-        WHERE ts_code IN {codes_tuple} AND trade_date BETWEEN '{start_date}' AND '{end_date}'
-        ORDER BY trade_date ASC
-        """
-        df_daily = pd.read_sql(daily_query, engine)
-        df_daily['trade_date'] = pd.to_datetime(df_daily['trade_date'])
-
-        # 3. 核心逻辑处理
-        final_report = []
-        for _, limit_row in df_limit.iterrows():
-            code = limit_row['ts_code']
-            stock_data = df_daily[df_daily['ts_code'] == code].copy()
-
-            if stock_data.empty: continue
-
-            # A. 找到区间内的最高价及其第一次出现的日期
-            max_high = float(stock_data['high'].max())
-            # 获取最高价所在的那一行（如果有多个相同最高价，取第一个）
-            max_high_row = stock_data[stock_data['high'] == max_high].iloc[0]
-            max_date = max_high_row['trade_date']
-
-            # B. 寻找最高价日期之后的最低价
-            post_max_data = stock_data[stock_data['trade_date'] > max_date]
-
-            if not post_max_data.empty:
-                min_low = float(post_max_data['low'].min())
-                last_close = float(post_max_data.iloc[-1]['close'])
-
-                price_diff = max_high - min_low
-                drop_pct = (price_diff / max_high * 100)
-                rebound_pct = ((last_close - min_low) / min_low * 100) if min_low > 0 else 0
-            else:
-                # 如果最高点出现在结束日当天，则没有“后期最低价”
-                min_low = None
-                price_diff = 0
-                drop_pct = 0
-                rebound_pct = 0
-
-            final_report.append({
-                "代码": code,
-                "名称": limit_row['name'],
-                "当日开盘": float(limit_row['limit_day_open'] or 0),
-                "当日最高": float(limit_row['limit_day_high'] or 0),
-                "区间最高": max_high,
-                "最高日期": max_date.strftime('%Y-%m-%d'),
-                "见顶后最低": min_low,
-                "最大回落幅%": round(drop_pct, 2),
-                "低位至期末波动%": round(rebound_pct, 2)
-            })
-
-        # 4. 排序与输出
-        if final_report:
-            result_df = pd.DataFrame(final_report)
-            result_df = result_df.sort_values(by="最大回落幅%", ascending=False)
-
-            print(f"\n分析区间: {start_date} 至 {end_date}")
-            print("说明: '见顶后最低' 指最高价日期之后的最低点")
-            print("=" * 120)
-            print(result_df.to_string(index=False, na_rep='N/A'))
-            print("=" * 120)
-
-    except Exception as e:
-        print(f"发生错误: {e}")
-
-
-# ==========================================
-# 设定日期
-# ==========================================
-if __name__ == "__main__":
-    START_INPUT = '2024-09-24'
-    END_INPUT = '2026-04-23'
-
-    analyze_stock_performance(START_INPUT, END_INPUT)

run_pipeline.py

@@ -0,0 +1,87 @@
+import argparse
+from datetime import datetime, timedelta
+
+from stock_config import TUSHARE_TOKEN
+
+
+def parse_yyyymmdd(value):
+    try:
+        return datetime.strptime(value, "%Y%m%d").date()
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("日期格式应为 YYYYMMDD") from exc
+
+
+def to_yyyymmdd(date_value):
+    return date_value.strftime("%Y%m%d")
+
+
+def to_iso_date(date_value):
+    return date_value.strftime("%Y-%m-%d")
+
+
+def run_stage(name, func, *args):
+    print(f"\n========== {name} 开始 ==========")
+    result = func(*args)
+    if result is False:
+        raise RuntimeError(f"{name} 执行失败，流程已停止")
+    print(f"========== {name} 完成 ==========")
+
+
+def main():
+    today = datetime.now().date()
+
+    parser = argparse.ArgumentParser(
+        description=(
+            "按 basic -> cal_date -> daily -> limit_up -> limit_list_d "
+            "-> second_board_profit_cache 顺序同步股票数据"
+        )
+    )
+    parser.add_argument(
+        "--date",
+        type=parse_yyyymmdd,
+        default=today,
+        help="同步单日数据，格式 YYYYMMDD；未指定时默认今天",
+    )
+    parser.add_argument("--start-date", type=parse_yyyymmdd, help="区间开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", type=parse_yyyymmdd, help="区间结束日期，格式 YYYYMMDD")
+    parser.add_argument("--skip-basic", action="store_true", help="跳过股票基础信息同步")
+    args = parser.parse_args()
+
+    start_date = args.start_date or args.date
+    end_date = args.end_date or args.date
+
+    if start_date > end_date:
+        raise SystemExit("开始日期不能大于结束日期")
+
+    start_yyyymmdd = to_yyyymmdd(start_date)
+    end_yyyymmdd = to_yyyymmdd(end_date)
+    limit_start = to_iso_date(start_date - timedelta(days=1))
+    limit_end = to_iso_date(end_date)
+
+    if not args.skip_basic:
+        from basic import sync_stock_basic
+
+        run_stage("basic 股票基础信息", sync_stock_basic)
+
+    from cal_date import run as run_cal_date
+    from daily import run as run_daily
+    from limit_list_d import run as run_limit_list_d
+    from limit_up import analyze_and_save_limits
+    from second_board_profit_cache import run as run_second_board_profit_cache
+
+    run_stage("cal_date 交易日历", run_cal_date, start_yyyymmdd, end_yyyymmdd, TUSHARE_TOKEN)
+    run_stage("daily 日行情", run_daily, start_yyyymmdd, end_yyyymmdd)
+    run_stage("limit_up 涨停/炸板", analyze_and_save_limits, limit_start, limit_end)
+    run_stage("limit_list_d 原始涨跌停数据", run_limit_list_d, start_yyyymmdd, end_yyyymmdd)
+    run_stage(
+        "second_board_profit_cache 二板获利缓存",
+        run_second_board_profit_cache,
+        start_yyyymmdd,
+        end_yyyymmdd,
+    )
+
+    print("\n全部流程执行完成")
+
+
+if __name__ == "__main__":
+    main()

second_board_profit_cache.py

@@ -0,0 +1,468 @@
+import argparse
+from datetime import datetime
+from decimal import Decimal, ROUND_HALF_UP
+
+from sqlalchemy import text
+
+from stock_config import get_engine
+
+
+DETAIL_TABLE = "second_board_profit_detail_cache"
+SUMMARY_TABLE = "second_board_profit_summary_cache"
+BATCH_SIZE = 1000
+
+
+def parse_date(value):
+    value = str(value).strip()
+    for fmt in ("%Y%m%d", "%Y-%m-%d"):
+        try:
+            return datetime.strptime(value, fmt).date()
+        except ValueError:
+            pass
+    raise ValueError(f"日期格式错误: {value}，应为 YYYYMMDD 或 YYYY-MM-DD")
+
+
+def to_decimal(value):
+    if value is None:
+        return None
+    return Decimal(str(value))
+
+
+def round2(value):
+    if value is None:
+        return None
+    return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
+
+
+def pct_change(value, base):
+    value = to_decimal(value)
+    base = to_decimal(base)
+    if value is None or base is None or base <= 0:
+        return None
+    return round2((value - base) / base * Decimal("100"))
+
+
+def format_limit_time(value):
+    text_value = "" if value is None else str(value).strip()
+    if not text_value or text_value.lower() in ("none", "null", "nan"):
+        return ""
+    if ":" in text_value:
+        return text_value
+    if len(text_value) == 4:
+        text_value = f"{text_value}00"
+    elif len(text_value) == 5:
+        text_value = f"0{text_value}"
+    elif len(text_value) != 6:
+        return ""
+    return f"{text_value[0:2]}:{text_value[2:4]}:{text_value[4:6]}"
+
+
+def time_seconds(value):
+    text_value = format_limit_time(value)
+    if not text_value:
+        return None
+
+    parts = text_value.split(":")
+    if len(parts) < 2 or len(parts) > 3:
+        return None
+
+    try:
+        hour = int(parts[0])
+        minute = int(parts[1])
+        second = int(parts[2]) if len(parts) == 3 else 0
+    except ValueError:
+        return None
+
+    if hour > 23 or minute > 59 or second > 59 or min(hour, minute, second) < 0:
+        return None
+    return hour * 3600 + minute * 60 + second
+
+
+def is_target_limit_time(value):
+    seconds = time_seconds(value)
+    if seconds is None:
+        return False
+    return 9 * 3600 + 31 * 60 <= seconds <= 9 * 3600 + 59 * 60 + 59
+
+
+def init_db(engine):
+    detail_sql = f"""
+    CREATE TABLE IF NOT EXISTS `{DETAIL_TABLE}` (
+        `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '自增主键',
+        `trade_date` DATE NOT NULL COMMENT 'T日交易日',
+        `previous_trade_date` DATE NOT NULL COMMENT 'T-1交易日',
+        `previous_two_trade_date` DATE NOT NULL COMMENT 'T-2交易日',
+        `next_trade_date` DATE NULL COMMENT 'T+1交易日',
+        `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码',
+        `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码',
+        `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称',
+        `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块',
+        `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业',
+        `previous_close` DECIMAL(16,4) NULL COMMENT 'T-1收盘价',
+        `limit_price` DECIMAL(16,4) NULL COMMENT 'T日涨停价',
+        `t_high` DECIMAL(16,4) NULL COMMENT 'T日最高价',
+        `t_close` DECIMAL(16,4) NULL COMMENT 'T日收盘价',
+        `t_pct_chg` DECIMAL(10,4) NULL COMMENT 'T日涨跌幅',
+        `status` VARCHAR(10) NOT NULL DEFAULT '' COMMENT 'limit封板 break炸板',
+        `first_limit_time` VARCHAR(8) NOT NULL DEFAULT '' COMMENT 'T日首次涨停时间',
+        `open_times` INT NULL COMMENT '开板次数',
+        `next_high` DECIMAL(16,4) NULL COMMENT 'T+1最高价',
+        `next_close` DECIMAL(16,4) NULL COMMENT 'T+1收盘价',
+        `next_intraday_profitable` TINYINT(1) NULL COMMENT 'T+1盘中是否盈利',
+        `next_high_profit_pct` DECIMAL(10,2) NULL COMMENT 'T+1最大获利百分比',
+        `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+        `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+        PRIMARY KEY (`id`),
+        UNIQUE KEY `uniq_trade_code` (`trade_date`, `ts_code`),
+        KEY `idx_ts_code_date` (`ts_code`, `trade_date`),
+        KEY `idx_trade_date` (`trade_date`),
+        KEY `idx_total_profit` (`next_high_profit_pct`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+    """
+    summary_sql = f"""
+    CREATE TABLE IF NOT EXISTS `{SUMMARY_TABLE}` (
+        `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码',
+        `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码',
+        `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称',
+        `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块',
+        `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业',
+        `second_board_count` INT NOT NULL DEFAULT 0 COMMENT '二板次数',
+        `limit_count` INT NOT NULL DEFAULT 0 COMMENT '封板次数',
+        `break_count` INT NOT NULL DEFAULT 0 COMMENT '炸板次数',
+        `profit_data_count` INT NOT NULL DEFAULT 0 COMMENT '具备收益数据次数',
+        `total_next_high_profit_pct` DECIMAL(12,2) NOT NULL DEFAULT 0 COMMENT '最大获利累计百分比',
+        `average_next_high_profit_pct` DECIMAL(10,2) NOT NULL DEFAULT 0 COMMENT '最大获利平均百分比',
+        `first_trade_date` DATE NULL COMMENT '首次命中日期',
+        `last_trade_date` DATE NULL COMMENT '最近命中日期',
+        `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+        PRIMARY KEY (`ts_code`),
+        KEY `idx_total_profit` (`total_next_high_profit_pct`),
+        KEY `idx_last_trade_date` (`last_trade_date`)
+    ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci
+    """
+    with engine.begin() as conn:
+        conn.execute(text(detail_sql))
+        conn.execute(text(summary_sql))
+    print(f"✅ 表 {DETAIL_TABLE} / {SUMMARY_TABLE} 检查完成。")
+
+
+def load_candidate_rows(conn, start_date, end_date):
+    query = text(
+        """
+        WITH calendar AS (
+            SELECT
+                cal_date AS trade_date,
+                LAG(cal_date) OVER (ORDER BY cal_date) AS previous_trade_date,
+                LAG(cal_date, 2) OVER (ORDER BY cal_date) AS previous_two_trade_date,
+                LEAD(cal_date) OVER (ORDER BY cal_date) AS next_trade_date
+            FROM trade_calendar
+            WHERE exchange = 'SSE' AND is_open = 1
+        ),
+        calendar_window AS (
+            SELECT *
+            FROM calendar
+            WHERE trade_date BETWEEN :start_date AND :end_date
+        )
+        SELECT
+            DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date,
+            DATE_FORMAT(previous_trade_date, '%Y-%m-%d') AS previous_trade_date,
+            DATE_FORMAT(previous_two_trade_date, '%Y-%m-%d') AS previous_two_trade_date,
+            COALESCE(DATE_FORMAT(next_trade_date, '%Y-%m-%d'), '') AS next_trade_date,
+            COALESCE(symbol, '') AS symbol,
+            ts_code,
+            name,
+            market,
+            industry,
+            previous_close,
+            limit_price,
+            t_high,
+            t_close,
+            t_pct_chg,
+            status,
+            first_limit_time,
+            open_times,
+            next_high,
+            next_close
+        FROM (
+            SELECT
+                c.trade_date,
+                c.previous_trade_date,
+                c.previous_two_trade_date,
+                c.next_trade_date,
+                d.symbol,
+                pl.ts_code,
+                COALESCE(NULLIF(sb.name, ''), NULLIF(pl.name, ''), '') AS name,
+                COALESCE(NULLIF(sb.market, ''), '') AS market,
+                COALESCE(NULLIF(sb.industry, ''), NULLIF(pl.industry, ''), '') AS industry,
+                COALESCE(d.pre_close, pl.close) AS previous_close,
+                ROUND(
+                    COALESCE(d.pre_close, pl.close) *
+                    CASE
+                        WHEN COALESCE(sb.market, '') = '北交所' THEN 1.30
+                        WHEN COALESCE(sb.market, '') IN ('创业板', '科创板') THEN 1.20
+                        WHEN COALESCE(sb.name, pl.name, '') LIKE '%ST%' THEN 1.05
+                        ELSE 1.10
+                    END,
+                    2
+                ) AS limit_price,
+                d.high AS t_high,
+                d.close AS t_close,
+                d.pct_chg AS t_pct_chg,
+                CASE
+                    WHEN ABS(d.close - d.high) < 0.0001 THEN 'limit'
+                    WHEN d.close < d.high THEN 'break'
+                    ELSE 'unknown'
+                END AS status,
+                CASE
+                    WHEN ABS(d.close - d.high) < 0.0001
+                        THEN NULLIF(NULLIF(TRIM(tlu.first_time), ''), 'None')
+                    WHEN d.close < d.high
+                        THEN NULLIF(NULLIF(TRIM(tlz.first_time), ''), 'None')
+                    ELSE NULL
+                END AS first_limit_time,
+                CASE
+                    WHEN ABS(d.close - d.high) < 0.0001 THEN tlu.open_times
+                    WHEN d.close < d.high THEN tlz.open_times
+                    ELSE NULL
+                END AS open_times,
+                nd.high AS next_high,
+                nd.close AS next_close
+            FROM calendar_window c
+            INNER JOIN limit_list_d pl
+                ON pl.trade_date = c.previous_trade_date AND pl.`limit` = 'U'
+            LEFT JOIN limit_list_d ppl
+                ON ppl.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+               AND ppl.trade_date = c.previous_two_trade_date
+               AND ppl.`limit` = 'U'
+            INNER JOIN stock_daily d
+                ON d.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+               AND d.trade_date = c.trade_date
+            LEFT JOIN stock_daily nd
+                ON nd.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+               AND nd.trade_date = c.next_trade_date
+            LEFT JOIN limit_list_d tlu
+                ON tlu.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+               AND tlu.trade_date = c.trade_date
+               AND tlu.`limit` = 'U'
+            LEFT JOIN limit_list_d tlz
+                ON tlz.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+               AND tlz.trade_date = c.trade_date
+               AND tlz.`limit` = 'Z'
+            LEFT JOIN stock_basic sb
+                ON sb.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci
+            WHERE c.previous_trade_date IS NOT NULL
+              AND c.previous_two_trade_date IS NOT NULL
+              AND ppl.ts_code IS NULL
+              AND d.high IS NOT NULL
+              AND d.close IS NOT NULL
+        ) candidate
+        WHERE previous_close IS NOT NULL
+          AND limit_price IS NOT NULL
+          AND limit_price > 0
+          AND t_high >= limit_price - 0.0001
+          AND status IN ('limit', 'break')
+        ORDER BY trade_date ASC, ts_code ASC
+        """
+    )
+    return [
+        dict(row)
+        for row in conn.execute(
+            query,
+            {"start_date": start_date, "end_date": end_date},
+        ).mappings()
+    ]
+
+
+def build_detail_records(rows):
+    records = []
+    for row in rows:
+        first_limit_time = format_limit_time(row["first_limit_time"])
+        if not is_target_limit_time(first_limit_time):
+            continue
+
+        next_intraday_profitable = None
+        if row["next_high"] is not None and row["t_high"] is not None:
+            next_intraday_profitable = (
+                1 if to_decimal(row["next_high"]) > to_decimal(row["t_high"]) else 0
+            )
+
+        records.append(
+            {
+                "trade_date": row["trade_date"],
+                "previous_trade_date": row["previous_trade_date"],
+                "previous_two_trade_date": row["previous_two_trade_date"],
+                "next_trade_date": row["next_trade_date"] or None,
+                "symbol": row["symbol"] or "",
+                "ts_code": row["ts_code"],
+                "name": row["name"] or "",
+                "market": row["market"] or "",
+                "industry": row["industry"] or "",
+                "previous_close": row["previous_close"],
+                "limit_price": row["limit_price"],
+                "t_high": row["t_high"],
+                "t_close": row["t_close"],
+                "t_pct_chg": row["t_pct_chg"],
+                "status": row["status"],
+                "first_limit_time": first_limit_time,
+                "open_times": row["open_times"],
+                "next_high": row["next_high"],
+                "next_close": row["next_close"],
+                "next_intraday_profitable": next_intraday_profitable,
+                "next_high_profit_pct": pct_change(row["next_high"], row["t_high"]),
+            }
+        )
+    return records
+
+
+def load_existing_keys(conn, start_date, end_date):
+    query = text(
+        f"""
+        SELECT DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date, ts_code
+        FROM `{DETAIL_TABLE}`
+        WHERE trade_date BETWEEN :start_date AND :end_date
+        """
+    )
+    rows = conn.execute(
+        query,
+        {"start_date": start_date, "end_date": end_date},
+    ).mappings()
+    return {(row["trade_date"], row["ts_code"]) for row in rows}
+
+
+def write_detail_records(conn, records, start_date, end_date):
+    existing_keys = load_existing_keys(conn, start_date, end_date)
+    overwritten_count = sum(
+        1 for record in records if (record["trade_date"], record["ts_code"]) in existing_keys
+    )
+    delete_result = conn.execute(
+        text(
+            f"""
+            DELETE FROM `{DETAIL_TABLE}`
+            WHERE trade_date BETWEEN :start_date AND :end_date
+            """
+        ),
+        {"start_date": start_date, "end_date": end_date},
+    )
+
+    if not records:
+        return 0, overwritten_count, delete_result.rowcount
+
+    sql = text(
+        f"""
+        INSERT INTO `{DETAIL_TABLE}` (
+            trade_date, previous_trade_date, previous_two_trade_date, next_trade_date,
+            symbol, ts_code, name, market, industry, previous_close, limit_price,
+            t_high, t_close, t_pct_chg, status, first_limit_time, open_times,
+            next_high, next_close, next_intraday_profitable, next_high_profit_pct
+        ) VALUES (
+            :trade_date, :previous_trade_date, :previous_two_trade_date, :next_trade_date,
+            :symbol, :ts_code, :name, :market, :industry, :previous_close, :limit_price,
+            :t_high, :t_close, :t_pct_chg, :status, :first_limit_time, :open_times,
+            :next_high, :next_close, :next_intraday_profitable, :next_high_profit_pct
+        )
+        ON DUPLICATE KEY UPDATE
+            previous_trade_date = VALUES(previous_trade_date),
+            previous_two_trade_date = VALUES(previous_two_trade_date),
+            next_trade_date = VALUES(next_trade_date),
+            symbol = VALUES(symbol),
+            name = VALUES(name),
+            market = VALUES(market),
+            industry = VALUES(industry),
+            previous_close = VALUES(previous_close),
+            limit_price = VALUES(limit_price),
+            t_high = VALUES(t_high),
+            t_close = VALUES(t_close),
+            t_pct_chg = VALUES(t_pct_chg),
+            status = VALUES(status),
+            first_limit_time = VALUES(first_limit_time),
+            open_times = VALUES(open_times),
+            next_high = VALUES(next_high),
+            next_close = VALUES(next_close),
+            next_intraday_profitable = VALUES(next_intraday_profitable),
+            next_high_profit_pct = VALUES(next_high_profit_pct)
+        """
+    )
+    for index in range(0, len(records), BATCH_SIZE):
+        conn.execute(sql, records[index:index + BATCH_SIZE])
+    return len(records), overwritten_count, delete_result.rowcount
+
+
+def refresh_summary(conn):
+    conn.execute(text(f"TRUNCATE TABLE `{SUMMARY_TABLE}`"))
+    conn.execute(
+        text(
+            f"""
+            INSERT INTO `{SUMMARY_TABLE}` (
+                ts_code, symbol, name, market, industry, second_board_count,
+                limit_count, break_count, profit_data_count, total_next_high_profit_pct,
+                average_next_high_profit_pct, first_trade_date, last_trade_date
+            )
+            SELECT
+                ts_code,
+                MAX(symbol) AS symbol,
+                MAX(name) AS name,
+                MAX(market) AS market,
+                MAX(industry) AS industry,
+                COUNT(*) AS second_board_count,
+                SUM(CASE WHEN status = 'limit' THEN 1 ELSE 0 END) AS limit_count,
+                SUM(CASE WHEN status = 'break' THEN 1 ELSE 0 END) AS break_count,
+                COUNT(next_high_profit_pct) AS profit_data_count,
+                ROUND(SUM(COALESCE(next_high_profit_pct, 0)), 2) AS total_next_high_profit_pct,
+                COALESCE(
+                    ROUND(
+                        SUM(COALESCE(next_high_profit_pct, 0))
+                        / NULLIF(COUNT(next_high_profit_pct), 0),
+                        2
+                    ),
+                    0
+                ) AS average_next_high_profit_pct,
+                MIN(trade_date) AS first_trade_date,
+                MAX(trade_date) AS last_trade_date
+            FROM `{DETAIL_TABLE}`
+            GROUP BY ts_code
+            """
+        )
+    )
+
+
+def run(start_date, end_date):
+    start = parse_date(start_date)
+    end = parse_date(end_date)
+    if start > end:
+        raise ValueError("开始日期不能大于结束日期")
+
+    engine = get_engine()
+    print(f"🚀 开始刷新二板获利缓存：{start} ~ {end}")
+
+    try:
+        init_db(engine)
+        with engine.begin() as conn:
+            rows = load_candidate_rows(conn, start, end)
+            records = build_detail_records(rows)
+            write_count, overwritten_count, deleted_count = write_detail_records(
+                conn,
+                records,
+                start,
+                end,
+            )
+            refresh_summary(conn)
+
+        print(f"候选记录：{len(rows)} 条")
+        print(f"命中记录：{len(records)} 条")
+        print(f"清理旧记录：{deleted_count} 条")
+        print(f"写入记录：{write_count} 条")
+        print(f"覆盖记录：{overwritten_count} 条")
+        print("✅ 二板获利缓存刷新完成。")
+        return True
+    except Exception as exc:
+        print(f"❌ 二板获利缓存刷新失败: {exc}")
+        return False
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="刷新二板获利缓存")
+    parser.add_argument("--start-date", default="20260515", help="开始日期，格式 YYYYMMDD 或 YYYY-MM-DD")
+    parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD 或 YYYY-MM-DD")
+    args = parser.parse_args()
+
+    run(args.start_date, args.end_date)

stock_config.py

@@ -0,0 +1,44 @@
+import os
+
+
+TUSHARE_TOKEN = os.getenv(
+    "TUSHARE_TOKEN",
+    "0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93",
+)
+
+# # 阿里云远程数据库
+# DB_CONFIG = {
+#     "host": os.getenv("MYSQL_HOST", "8.159.129.156"),
+#     "port": int(os.getenv("MYSQL_PORT", "10836")),
+#     "user": os.getenv("MYSQL_USER", "yangfan"),
+#     "password": os.getenv("MYSQL_PASSWORD", "aA%8023321088"),
+#     "database": os.getenv("MYSQL_DATABASE", "level"),
+#     "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),
+# }
+
+# mac本地mysql
+DB_CONFIG = {
+    "host": os.getenv("MYSQL_HOST", "localhost"),
+    "port": int(os.getenv("MYSQL_PORT", "3306")),
+    "user": os.getenv("MYSQL_USER", "yangfan"),
+    "password": os.getenv("MYSQL_PASSWORD", "admin"),
+    "database": os.getenv("MYSQL_DATABASE", "level"),
+    "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),
+}
+
+
+def get_db_url():
+    return (
+        f"mysql+pymysql://{DB_CONFIG['user']}@{DB_CONFIG['host']}:"
+        f"{DB_CONFIG['port']}/{DB_CONFIG['database']}?charset={DB_CONFIG['charset']}"
+    )
+
+
+def get_engine():
+    from sqlalchemy import create_engine
+
+    return create_engine(
+        get_db_url(),
+        connect_args={"password": DB_CONFIG["password"]},
+        pool_pre_ping=True,
+    )