修改涨停板逻辑

limit_up.py

@@ -3,6 +3,8 @@ from sqlalchemy import 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
 
@@ -11,9 +13,29 @@ 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"
 
 
+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})
 
@@ -52,7 +74,6 @@ def get_combined_table_sql():
         `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前收盘价',
-        `vol` BIGINT COMMENT 'stock_daily成交量',
         `calc_vol` BIGINT COMMENT 'stock_daily成交量',
         `market` VARCHAR(20) COMMENT '所属板块',
         `tushare_limit_type` VARCHAR(1) COMMENT 'Tushare-D跌停U涨停Z炸板',
@@ -101,20 +122,47 @@ def index_exists(conn, table_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 = {
-        "vol": "BIGINT COMMENT 'stock_daily成交量'",
         "calc_vol": "BIGINT COMMENT 'stock_daily成交量'",
     }
 
     for column_name, column_def in columns.items():
         if column_name not in existing_columns:
-            after_column = "calc_pre_close" if column_name == "vol" else "vol"
             conn.execute(
                 text(
                     f"ALTER TABLE `{COMBINED_TABLE}` "
-                    f"ADD COLUMN `{column_name}` {column_def} AFTER `{after_column}`"
+                    f"ADD COLUMN `{column_name}` {column_def} AFTER `calc_pre_close`"
                 )
             )
 
@@ -128,7 +176,6 @@ def sync_combined_vol_from_daily(conn):
               ON c.trade_date = d.trade_date
              AND c.ts_code COLLATE utf8mb4_unicode_ci = d.ts_code COLLATE utf8mb4_unicode_ci
             SET c.symbol = d.symbol,
-                c.vol = d.vol,
                 c.calc_vol = d.vol
             """
         )
@@ -161,26 +208,26 @@ def dedupe_combined_table(conn):
                     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.vol IS NOT NULL OR c1.calc_vol IS NOT NULL, 1, 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.vol IS NOT NULL OR c2.calc_vol IS NOT NULL, 1, 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.vol IS NOT NULL OR c1.calc_vol IS NOT NULL, 1, 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.vol IS NOT NULL OR c2.calc_vol IS NOT NULL, 1, 0)
+                        + IF(c2.calc_vol IS NOT NULL, 1, 0)
                     )
                     AND c1.id < c2.id
                 )
@@ -191,7 +238,14 @@ def dedupe_combined_table(conn):
 
 def ensure_combined_indexes(conn):
     dedupe_combined_table(conn)
-    if not index_exists(conn, COMBINED_TABLE, "uniq_date_code"):
+    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
+    ):
+        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
+    ):
         conn.execute(
             text(
                 f"ALTER TABLE `{COMBINED_TABLE}` "
@@ -219,6 +273,38 @@ def optimize_combined_table(conn):
     ensure_combined_columns(conn)
     sync_combined_vol_from_daily(conn)
     ensure_combined_indexes(conn)
+    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):
@@ -228,20 +314,31 @@ def fetch_tushare_limit_list(trade_dates):
         "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,
@@ -253,19 +350,24 @@ def fetch_tushare_limit_list(trade_dates):
                 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"])
 
-    return pd.concat(frames, ignore_index=True).drop_duplicates(
+    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):
@@ -437,7 +539,6 @@ def build_combined_limit_data(daily_df, calc_df, tushare_df):
     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["vol"] = combined_df["vol"]
     combined_df["calc_vol"] = combined_df["vol"]
 
     combined_df["final_status"] = combined_df.apply(resolve_final_status, axis=1)
@@ -460,7 +561,6 @@ def build_combined_limit_data(daily_df, calc_df, tushare_df):
         "calc_high",
         "calc_close",
         "calc_pre_close",
-        "vol",
         "calc_vol",
         "market",
         *tushare_columns,
@@ -496,7 +596,10 @@ def analyze_and_save_limits(start_date, end_date):
     engine = get_engine()
 
     # 0. 初始化检查表结构
+    print(f"🚀 开始处理涨停/炸板数据：({start_date} -> {end_date}]")
+    print(f"🧱 初始化并检查融合表 {COMBINED_TABLE} ...")
     init_db(engine)
+    print("✅ 表结构、历史重复数据和唯一索引检查完成。")
 
     requested_start_date = pd.to_datetime(start_date).date()
     if requested_start_date < LOCAL_START_DATE:
@@ -518,6 +621,7 @@ def analyze_and_save_limits(start_date, end_date):
     """)
 
     try:
+        print("📥 读取 stock_daily 与 stock_basic 数据 ...")
         df = pd.read_sql(
             query,
             engine,
@@ -530,6 +634,8 @@ def analyze_and_save_limits(start_date, end_date):
         if df.empty:
             print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。")
             return True
+        target_dates = df['trade_date'].astype(str).unique()
+        print(f"✅ 读取完成：{len(df)} 行行情，覆盖 {len(target_dates)} 个交易日。")
 
         # 2. 涨停价计算规则
         def calc_limit(row):
@@ -551,10 +657,16 @@ 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)} 条。"
+        )
 
-        target_dates = df['trade_date'].astype(str).unique()
         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 = {
@@ -564,9 +676,12 @@ def analyze_and_save_limits(start_date, end_date):
             combined_delete_query = text(
                 f"DELETE FROM {COMBINED_TABLE} WHERE trade_date IN ({placeholders})"
             )
-            conn.execute(combined_delete_query, delete_params)
+            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,
@@ -575,6 +690,10 @@ def analyze_and_save_limits(start_date, end_date):
                     chunksize=1000,
                     method="multi",
                 )
+            else:
+                print("ℹ️ 融合结果为空，本次不写入新数据。")
+            validate_no_duplicate_combined_rows(conn)
+            print("✅ 写入后数据库重复键校验通过。")
 
         if not save_df.empty:
             print(f"✅ 处理成功！涉及交易日: {len(target_dates)} 天")
@@ -593,7 +712,7 @@ def cleanup_combined_data():
     engine = get_engine()
     try:
         init_db(engine)
-        print(f"✅ {COMBINED_TABLE} 已完成历史重复数据清理、唯一索引检查和 vol 回填。")
+        print(f"✅ {COMBINED_TABLE} 已完成历史重复数据清理、唯一索引检查和 calc_vol 回填。")
         return True
     except Exception as e:
         print(f"❌ 清理失败: {e}")
@@ -605,12 +724,12 @@ def cleanup_combined_data():
 # ============================================================
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据")
-    parser.add_argument("--start-date", default="2026-03-02", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
+    parser.add_argument("--start-date", default="2020-01-02", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
-    parser.add_argument("--end-date", default="2026-04-30", help="结束日期，格式 YYYY-MM-DD")
+    parser.add_argument("--end-date", default="2024-12-31", help="结束日期，格式 YYYY-MM-DD")
     parser.add_argument(
         "--cleanup-only",
         action="store_true",
-        help="仅清理 stock_limit_combined_data 历史重复数据并从 stock_daily 回填 vol",
+        help="仅清理 stock_limit_combined_data 历史重复数据并从 stock_daily 回填 calc_vol",
     )
     args = parser.parse_args()