新增涨跌停炸板单独表脚本

cal_date.py

@@ -131,7 +131,7 @@ def run(start_date, end_date, token):
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="同步交易日历")
     parser.add_argument("--start-date", default="20260429", help="开始日期，格式 YYYYMMDD")
-    parser.add_argument("--end-date", default="20260507", help="结束日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260508", help="结束日期，格式 YYYYMMDD")
     args = parser.parse_args()
 
     run(args.start_date, args.end_date, TUSHARE_TOKEN)

daily.py

@@ -157,8 +157,8 @@ def run(start_date, end_date):
 # ===== 输入参数 =====
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="同步日行情")
-    parser.add_argument("--start-date", default="20260506", help="开始日期，格式 YYYYMMDD")
+    parser.add_argument("--start-date", default="20260430", help="开始日期，格式 YYYYMMDD")
-    parser.add_argument("--end-date", default="20260507", help="结束日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260508", help="结束日期，格式 YYYYMMDD")
     args = parser.parse_args()
 
     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 = "20200101"  # 开始日期，格式 YYYYMMDD 或 YYYY-MM-DD
+    END_DATE = "20260508"  # 结束日期，格式 YYYYMMDD 或 YYYY-MM-DD
+
+    run(START_DATE, END_DATE)

limit_up.py

@@ -786,8 +786,8 @@ def cleanup_combined_data():
 # ============================================================
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据")
-    parser.add_argument("--start-date", default="2026-04-30", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
+    parser.add_argument("--start-date", default="2019-12-30", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
-    parser.add_argument("--end-date", default="2026-05-06", help="结束日期，格式 YYYY-MM-DD")
+    parser.add_argument("--end-date", default="2026-05-08", help="结束日期，格式 YYYY-MM-DD")
     parser.add_argument(
         "--cleanup-only",
         action="store_true",

run_pipeline.py

@@ -31,7 +31,7 @@ def main():
     today = datetime.now().date()
 
     parser = argparse.ArgumentParser(
-        description="按 basic -> cal_date -> daily -> limit_up 顺序同步股票数据"
+        description="按 basic -> cal_date -> daily -> limit_up -> limit_list_d 顺序同步股票数据"
     )
     parser.add_argument(
         "--date",
@@ -62,11 +62,13 @@ def main():
 
     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
 
     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)
 
     print("\n全部流程执行完成")