diff --git a/cal_date.py b/cal_date.py index a82808a..014871c 100644 --- a/cal_date.py +++ b/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) diff --git a/daily.py b/daily.py index ea62777..6e9c6bc 100644 --- a/daily.py +++ b/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("--end-date", default="20260507", help="结束日期，格式 YYYYMMDD") + parser.add_argument("--start-date", default="20260430", help="开始日期，格式 YYYYMMDD") + parser.add_argument("--end-date", default="20260508", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() run(args.start_date, args.end_date) diff --git a/limit_list_d.py b/limit_list_d.py new file mode 100644 index 0000000..2fcb5e5 --- /dev/null +++ b/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) diff --git a/limit_up.py b/limit_up.py index 032eb65..d5cc3ef 100644 --- a/limit_up.py +++ b/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("--end-date", default="2026-05-06", 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-08", help="结束日期，格式 YYYY-MM-DD") parser.add_argument( "--cleanup-only", action="store_true", diff --git a/run_pipeline.py b/run_pipeline.py index 48dc85f..53e30b4 100644 --- a/run_pipeline.py +++ b/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全部流程执行完成")