From f8395f706eaa229cbff70b14bcdd9a84182c7c6c Mon Sep 17 00:00:00 2001 From: yangfan <784310601@qq.com> Date: Tue, 5 May 2026 14:13:11 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E6=94=B9limit=5Fup=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E6=9D=A5=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cal_date.py | 2 +- daily.py | 29 ++- limit_up.py | 681 +++++++++++++++++++++++++++++++++--------------- stock_config.py | 17 +- 4 files changed, 507 insertions(+), 222 deletions(-) diff --git a/cal_date.py b/cal_date.py index 962cae3..43b87f2 100644 --- a/cal_date.py +++ b/cal_date.py @@ -130,7 +130,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("--start-date", default="20180101", help="开始日期，格式 YYYYMMDD") parser.add_argument("--end-date", default="20260430", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() diff --git a/daily.py b/daily.py index ee60c33..9e8a5a1 100644 --- a/daily.py +++ b/daily.py @@ -1,5 +1,6 @@ import tushare as ts import pymysql +import pandas as pd from datetime import datetime import time import argparse @@ -53,6 +54,14 @@ 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() @@ -90,15 +99,15 @@ def insert_data(df): ts_code, symbol, str_to_date(row['trade_date']), # 保持 DATE 类型 - row['open'], - row['high'], - row['low'], - row['close'], - row['pre_close'], - row['change'], - row['pct_chg'], - row['vol'], - row['amount'] + mysql_value(row['open']), + mysql_value(row['high']), + mysql_value(row['low']), + mysql_value(row['close']), + mysql_value(row['pre_close']), + mysql_value(row['change']), + mysql_value(row['pct_chg']), + mysql_value(row['vol']), + mysql_value(row['amount']) )) db_cursor.executemany(sql, data) @@ -148,7 +157,7 @@ def run(start_date, end_date): # ===== 输入参数 ===== if __name__ == "__main__": parser = argparse.ArgumentParser(description="同步日行情") - parser.add_argument("--start-date", default="20260430", help="开始日期，格式 YYYYMMDD") + parser.add_argument("--start-date", default="20180102", help="开始日期，格式 YYYYMMDD") parser.add_argument("--end-date", default="20260430", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() diff --git a/limit_up.py b/limit_up.py index 9e6ec8e..48776d2 100644 --- a/limit_up.py +++ b/limit_up.py @@ -1,202 +1,333 @@ import pandas as pd from sqlalchemy import text import argparse -import re +import tushare as ts +from datetime import date -from stock_config import get_engine +from stock_config import TUSHARE_TOKEN, get_engine # --- 数据库配置 --- # 请根据实际情况修改数据库连接信息 # DB_CONN_STR = 'mysql+pymysql://root:password@127.0.0.1:3306/stock_data' -AK_START_DATE = pd.to_datetime("2026-04-01").date() +LOCAL_START_DATE = date(2018, 1, 1) +TUSHARE_LIMIT_LIST_START_DATE = date(2020, 1, 1) +COMBINED_TABLE = "stock_limit_combined_data" -def normalize_stock_name(value): - if pd.isna(value): - return "" +def normalize_trade_dates(trade_dates): + return sorted({pd.to_datetime(trade_date).date() for trade_date in trade_dates}) - text_value = str(value).upper() - text_value = re.sub(r"\s+", "", text_value) - text_value = re.sub(r"[\*＊·・.。,_，、/\\()（）\[\]【】\-—_]+", "", text_value) - for marker in ["行权", "除权", "复权"]: - text_value = text_value.replace(marker, "") - - return text_value +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): """ 初始化数据库表结构 """ - create_table_sql = """ - CREATE TABLE IF NOT EXISTS `stock_limit_data` ( + with engine.begin() as conn: + conn.execute(text(get_combined_table_sql())) + optimize_combined_table(conn) + + +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 '股票代码', - `ts_code` VARCHAR(20) NOT NULL COMMENT 'TS股票代码', - `name` VARCHAR(50) NOT NULL COMMENT '股票名称', - `status` TINYINT NOT NULL COMMENT '状态：1-涨停，2-炸板', - `limit_price` DECIMAL(16, 4) COMMENT '计算出的理论涨停价', - `close` DECIMAL(16, 4) COMMENT '实际收盘价', - `vol` BIGINT COMMENT '成交量', + `symbol` VARCHAR(10) NOT NULL COMMENT '股票代码，以stock_daily为准', + `ts_code` VARCHAR(20) NOT NULL COMMENT 'TS股票代码，以stock_daily为准', + `name` VARCHAR(50) NOT NULL COMMENT '股票名称，以stock_daily关联基础表为准', + `final_status` TINYINT NOT NULL COMMENT '融合状态：1-涨停，2-炸板', + `source_count` INT NOT NULL DEFAULT 0 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前收盘价', + `vol` BIGINT COMMENT 'stock_daily成交量', + `calc_vol` BIGINT COMMENT 'stock_daily成交量', `market` VARCHAR(20) COMMENT '所属板块', - `ak_code` VARCHAR(10) COMMENT 'AKShare-代码', - `ak_name` VARCHAR(50) COMMENT 'AKShare-名称', - `ak_pct_chg` DECIMAL(10, 4) COMMENT 'AKShare-涨跌幅', - `ak_latest_price` DECIMAL(16, 4) COMMENT 'AKShare-最新价', - `ak_amount` DECIMAL(24, 4) COMMENT 'AKShare-成交额', - `ak_float_market_value` DECIMAL(24, 4) COMMENT 'AKShare-流通市值', - `ak_total_market_value` DECIMAL(24, 4) COMMENT 'AKShare-总市值', - `ak_turnover_rate` DECIMAL(10, 4) COMMENT 'AKShare-换手率', - `ak_seal_amount` DECIMAL(24, 4) COMMENT 'AKShare-封板资金', - `ak_first_limit_time` VARCHAR(20) COMMENT 'AKShare-首次封板时间', - `ak_last_limit_time` VARCHAR(20) COMMENT 'AKShare-最后封板时间', - `ak_break_limit_count` INT COMMENT 'AKShare-炸板次数', - `ak_limit_up_stat` VARCHAR(50) COMMENT 'AKShare-涨停统计', - `ak_limit_up_days` INT COMMENT 'AKShare-连板数', - `ak_industry` VARCHAR(100) COMMENT 'AKShare-所属行业', - UNIQUE KEY `uniq_date_code` (`trade_date`, `ts_code`) - ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4; + `tushare_limit_type` VARCHAR(1) COMMENT 'Tushare-D跌停U涨停Z炸板', + `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)) - ensure_base_columns(conn) - ensure_ak_columns(conn) -def ensure_base_columns(conn): - existing_columns = { +def get_table_columns(conn, table_name): + return { row[0] - for row in conn.execute(text("SHOW COLUMNS FROM `stock_limit_data`")).fetchall() - } - base_columns = { - "vol": "BIGINT COMMENT '成交量'", + for row in conn.execute(text(f"SHOW COLUMNS FROM `{table_name}`")).fetchall() } - for column_name, column_def in base_columns.items(): + +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 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: - conn.execute(text(f"ALTER TABLE `stock_limit_data` ADD COLUMN `{column_name}` {column_def}")) + 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}`" + ) + ) -def ensure_ak_columns(conn): - existing_columns = { - row[0] - for row in conn.execute(text("SHOW COLUMNS FROM `stock_limit_data`")).fetchall() - } - ak_columns = { - "ak_code": "VARCHAR(10) COMMENT 'AKShare-代码'", - "ak_name": "VARCHAR(50) COMMENT 'AKShare-名称'", - "ak_pct_chg": "DECIMAL(10, 4) COMMENT 'AKShare-涨跌幅'", - "ak_latest_price": "DECIMAL(16, 4) COMMENT 'AKShare-最新价'", - "ak_amount": "DECIMAL(24, 4) COMMENT 'AKShare-成交额'", - "ak_float_market_value": "DECIMAL(24, 4) COMMENT 'AKShare-流通市值'", - "ak_total_market_value": "DECIMAL(24, 4) COMMENT 'AKShare-总市值'", - "ak_turnover_rate": "DECIMAL(10, 4) COMMENT 'AKShare-换手率'", - "ak_seal_amount": "DECIMAL(24, 4) COMMENT 'AKShare-封板资金'", - "ak_first_limit_time": "VARCHAR(20) COMMENT 'AKShare-首次封板时间'", - "ak_last_limit_time": "VARCHAR(20) COMMENT 'AKShare-最后封板时间'", - "ak_break_limit_count": "INT COMMENT 'AKShare-炸板次数'", - "ak_limit_up_stat": "VARCHAR(50) COMMENT 'AKShare-涨停统计'", - "ak_limit_up_days": "INT COMMENT 'AKShare-连板数'", - "ak_industry": "VARCHAR(100) COMMENT 'AKShare-所属行业'", - } - - for column_name, column_def in ak_columns.items(): - if column_name not in existing_columns: - conn.execute(text(f"ALTER TABLE `stock_limit_data` ADD COLUMN `{column_name}` {column_def}")) +def sync_combined_vol_from_daily(conn): + conn.execute( + text( + f""" + UPDATE `{COMBINED_TABLE}` c + JOIN `stock_daily` d + 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 + """ + ) + ) -def fetch_ak_limit_pool(trade_dates): - try: - import akshare as ak - except ImportError as exc: - raise RuntimeError("缺少 akshare 依赖，请先安装: pip install akshare") from exc +def dedupe_combined_table(conn): + conn.execute( + text( + f""" + DELETE c + FROM `{COMBINED_TABLE}` c + LEFT JOIN `stock_daily` d + ON c.trade_date = d.trade_date + AND c.ts_code COLLATE utf8mb4_unicode_ci = d.ts_code COLLATE utf8mb4_unicode_ci + 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 COLLATE utf8mb4_unicode_ci = c2.ts_code COLLATE utf8mb4_unicode_ci + 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.vol IS NOT NULL OR 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) + ) + 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) + ) = + ( + 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) + ) + AND c1.id < c2.id + ) + """ + ) + ) + + +def ensure_combined_indexes(conn): + dedupe_combined_table(conn) + if not index_exists(conn, COMBINED_TABLE, "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"): + conn.execute( + text(f"ALTER TABLE `{COMBINED_TABLE}` ADD KEY `idx_trade_date` (`trade_date`)") + ) + if not index_exists(conn, COMBINED_TABLE, "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"): + conn.execute( + text( + f"ALTER TABLE `{COMBINED_TABLE}` " + "ADD KEY `idx_trade_date_status` (`trade_date`, `final_status`)" + ) + ) + + +def optimize_combined_table(conn): + ensure_combined_columns(conn) + sync_combined_vol_from_daily(conn) + ensure_combined_indexes(conn) + + +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: + return pd.DataFrame(columns=["trade_date", "symbol", "ts_code"]) + + ts.set_token(TUSHARE_TOKEN) + pro = ts.pro_api() frames = [] - for trade_date in trade_dates: - trade_day = pd.to_datetime(trade_date).date() - if trade_day < AK_START_DATE: - continue + 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" + ) + for trade_date in tushare_trade_dates: + tushare_date = pd.to_datetime(trade_date).strftime("%Y%m%d") + for limit_type in ("U", "Z"): + 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 - ak_date = pd.to_datetime(trade_date).strftime("%Y%m%d") - try: - ak_df = ak.stock_zt_pool_em(date=ak_date) - except Exception as exc: - print(f"⚠️ AKShare 涨停池获取失败: {ak_date}, 错误: {exc}") - continue + if ts_df is None or ts_df.empty: + continue - if ak_df is None or ak_df.empty: - continue - - normalized_df = normalize_ak_limit_pool(ak_df, trade_date) - if not normalized_df.empty: - frames.append(normalized_df) + normalized_df = normalize_tushare_limit_list(ts_df) + if not normalized_df.empty: + frames.append(normalized_df) if not frames: - return pd.DataFrame(columns=["trade_date", "symbol", "name"]) + return pd.DataFrame(columns=["trade_date", "symbol", "ts_code"]) - return pd.concat(frames, ignore_index=True) + return pd.concat(frames, ignore_index=True).drop_duplicates( + subset=["trade_date", "ts_code", "tushare_limit_type"], + keep="first", + ) -def normalize_ak_limit_pool(ak_df, trade_date): +def normalize_tushare_limit_list(ts_df): column_map = { - "代码": "ak_code", - "名称": "ak_name", - "涨跌幅": "ak_pct_chg", - "最新价": "ak_latest_price", - "成交额": "ak_amount", - "流通市值": "ak_float_market_value", - "总市值": "ak_total_market_value", - "换手率": "ak_turnover_rate", - "封板资金": "ak_seal_amount", - "首次封板时间": "ak_first_limit_time", - "最后封板时间": "ak_last_limit_time", - "炸板次数": "ak_break_limit_count", - "涨停统计": "ak_limit_up_stat", - "连板数": "ak_limit_up_days", - "所属行业": "ak_industry", + "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", } - missing_columns = [column for column in column_map if column not in ak_df.columns] - if missing_columns: - print(f"⚠️ AKShare 返回字段缺失，已跳过缺失字段: {', '.join(missing_columns)}") - - if "代码" not in ak_df.columns or "名称" not in ak_df.columns: - return pd.DataFrame(columns=["trade_date", "symbol", "name"]) + 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 ak_df.columns + source: target for source, target in column_map.items() if source in ts_df.columns } - normalized_df = ak_df[list(available_map)].rename(columns=available_map).copy() - normalized_df["trade_date"] = pd.to_datetime(trade_date).date() - normalized_df["symbol"] = ( - normalized_df["ak_code"].astype(str).str.extract(r"(\d+)", expand=False).str.zfill(6) - ) - normalized_df["name"] = normalized_df["ak_name"].astype(str).str.strip() - normalized_df["normalized_name"] = normalized_df["name"].apply(normalize_stock_name) + 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 [ - "ak_pct_chg", - "ak_latest_price", - "ak_amount", - "ak_float_market_value", - "ak_total_market_value", - "ak_turnover_rate", - "ak_seal_amount", - "ak_break_limit_count", - "ak_limit_up_days", + "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 [ - "ak_code", - "ak_name", - "ak_first_limit_time", - "ak_last_limit_time", - "ak_limit_up_stat", - "ak_industry", + "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) @@ -204,54 +335,158 @@ def normalize_ak_limit_pool(ak_df, trade_date): return normalized_df -def merge_ak_limit_pool(final_df, ak_df): - ak_columns = [ - "ak_code", - "ak_name", - "ak_pct_chg", - "ak_latest_price", - "ak_amount", - "ak_float_market_value", - "ak_total_market_value", - "ak_turnover_rate", - "ak_seal_amount", - "ak_first_limit_time", - "ak_last_limit_time", - "ak_break_limit_count", - "ak_limit_up_stat", - "ak_limit_up_days", - "ak_industry", +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 ak_columns: - if column not in ak_df.columns: - ak_df[column] = None - final_df["normalized_name"] = final_df["name"].apply(normalize_stock_name) - if "normalized_name" not in ak_df.columns: - ak_df["normalized_name"] = ak_df["name"].apply(normalize_stock_name) + for column in tushare_columns: + if column not in tushare_df.columns: + tushare_df[column] = None - merge_keys = ["trade_date", "symbol"] - ak_merge_df = ak_df[ - merge_keys + ["normalized_name"] + ak_columns - ].drop_duplicates(subset=merge_keys) - merged_df = final_df.merge( - ak_merge_df, - how="left", - on=merge_keys, - suffixes=("", "_ak_match"), + 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"], ) - name_mismatch = ( - merged_df["ak_code"].notna() - & merged_df["normalized_name_ak_match"].notna() - & (merged_df["normalized_name"] != merged_df["normalized_name_ak_match"]) - ) - if name_mismatch.any(): - print(f"⚠️ 有 {int(name_mismatch.sum())} 条记录代码一致但名称不一致，已按代码合并 AKShare 字段。") + if combined_df.empty: + return pd.DataFrame() - return merged_df[[ - "trade_date", "symbol", "ts_code", "name", "status", "limit_price", "close", "vol", "market", - *ak_columns, - ]] + 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["vol"] = combined_df["vol"] + 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", + "vol", + "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): @@ -263,15 +498,21 @@ def analyze_and_save_limits(start_date, end_date): # 0. 初始化检查表结构 init_db(engine) + requested_start_date = pd.to_datetime(start_date).date() + 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, d.vol, b.market FROM stock_daily d - JOIN stock_basic b ON d.ts_code = b.ts_code + JOIN stock_basic b + ON d.ts_code COLLATE utf8mb4_unicode_ci = b.ts_code COLLATE utf8mb4_unicode_ci 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 """) @@ -280,7 +521,11 @@ def analyze_and_save_limits(start_date, end_date): df = pd.read_sql( query, engine, - params={"start_date": start_date, "end_date": end_date}, + params={ + "start_date": start_date, + "end_date": end_date, + "local_start_date": LOCAL_START_DATE, + }, ) if df.empty: print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。") @@ -308,35 +553,35 @@ def analyze_and_save_limits(start_date, end_date): save_df = pd.concat([limit_df, broken_df]) target_dates = df['trade_date'].astype(str).unique() - ak_df = fetch_ak_limit_pool(target_dates) + tushare_df = fetch_tushare_limit_list(target_dates) + combined_df = build_combined_limit_data(df.copy(), save_df.copy(), tushare_df.copy()) 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))) - delete_query = text( - f"DELETE FROM stock_limit_data WHERE trade_date IN ({placeholders})" + combined_delete_query = text( + f"DELETE FROM {COMBINED_TABLE} WHERE trade_date IN ({placeholders})" ) - conn.execute(delete_query, delete_params) + conn.execute(combined_delete_query, delete_params) - if not save_df.empty: - final_df = save_df[[ - 'trade_date', 'symbol', 'ts_code', 'name', - 'status', 'limit_price', 'close', 'vol', 'market' - ]].copy() - final_df["trade_date"] = pd.to_datetime(final_df["trade_date"]).dt.date - final_df["symbol"] = final_df["symbol"].astype(str).str.zfill(6) - final_df["name"] = final_df["name"].astype(str).str.strip() - final_df = merge_ak_limit_pool(final_df, ak_df) - final_df.to_sql('stock_limit_data', conn, if_exists='append', index=False) + if not combined_df.empty: + combined_df.to_sql( + COMBINED_TABLE, + conn, + if_exists='append', + index=False, + chunksize=1000, + method="multi", + ) if not save_df.empty: - matched_count = int(final_df["ak_code"].notna().sum()) print(f"✅ 处理成功！涉及交易日: {len(target_dates)} 天") - print(f"📊 总计入库: {len(final_df)} 条记录，AKShare 匹配补充: {matched_count} 条。") + print(f"📊 融合表 {COMBINED_TABLE} 入库: {len(combined_df)} 条记录。") else: - print(f"ℹ️ 该时段内全市场无涨停或炸板记录，已清理 {len(target_dates)} 个交易日的旧结果。") + print(f"ℹ️ 该时段内本地行情无涨停或炸板记录，已清理 {len(target_dates)} 个交易日的融合表旧结果。") + print(f"📊 融合表 {COMBINED_TABLE} 入库: {len(combined_df)} 条记录。") return True except Exception as e: @@ -344,15 +589,35 @@ def analyze_and_save_limits(start_date, end_date): return False +def cleanup_combined_data(): + engine = get_engine() + try: + init_db(engine) + print(f"✅ {COMBINED_TABLE} 已完成历史重复数据清理、唯一索引检查和 vol 回填。") + return True + except Exception as e: + print(f"❌ 清理失败: {e}") + return False + + # ============================================================ # 用户配置区 # ============================================================ if __name__ == "__main__": parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据") - parser.add_argument("--start-date", default="2026-04-29", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") + 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( + "--cleanup-only", + action="store_true", + help="仅清理 stock_limit_combined_data 历史重复数据并从 stock_daily 回填 vol", + ) args = parser.parse_args() + if args.cleanup_only: + cleanup_combined_data() + raise SystemExit(0) + # 开始日期：计算从该日期的下一个交易日开始 START_DATE = args.start_date diff --git a/stock_config.py b/stock_config.py index 1790a6f..baf6b27 100644 --- a/stock_config.py +++ b/stock_config.py @@ -6,11 +6,22 @@ TUSHARE_TOKEN = os.getenv( "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", "8.159.129.156"), - "port": int(os.getenv("MYSQL_PORT", "10836")), + "host": os.getenv("MYSQL_HOST", "localhost"), + "port": int(os.getenv("MYSQL_PORT", "3306")), "user": os.getenv("MYSQL_USER", "yangfan"), - "password": os.getenv("MYSQL_PASSWORD", "aA%8023321088"), + "password": os.getenv("MYSQL_PASSWORD", "admin"), "database": os.getenv("MYSQL_DATABASE", "level"), "charset": os.getenv("MYSQL_CHARSET", "utf8mb4"), }