From ea3d96905622f73939eed5e5bc8a4ab141c18ed8 Mon Sep 17 00:00:00 2001 From: mianyang <784310601@qq.com> Date: Tue, 5 May 2026 19:05:47 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9=E6=B6=A8=E5=81=9C=E6=9D=BF?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- limit_up.py | 157 +++++++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 138 insertions(+), 19 deletions(-) diff --git a/limit_up.py b/limit_up.py index 48776d2..292c077 100644 --- a/limit_up.py +++ b/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("--end-date", default="2026-04-30", help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--start-date", default="2020-01-02", 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()