From de5da361e84e43c2fa0095f4872a9dfb56f8d714 Mon Sep 17 00:00:00 2001 From: mianyang <784310601@qq.com> Date: Sun, 17 May 2026 13:19:57 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E8=8E=B7=E5=8F=96=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E4=BA=8C=E6=9D=BF=E6=95=B0=E6=8D=AE=E8=84=9A=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cal_date.py | 4 +- daily.py | 4 +- limit_list_d.py | 4 +- limit_up.py | 4 +- run_pipeline.py | 12 +- second_board_profit_cache.py | 468 +++++++++++++++++++++++++++++++++++ 6 files changed, 487 insertions(+), 9 deletions(-) create mode 100644 second_board_profit_cache.py diff --git a/cal_date.py b/cal_date.py index a75c793..0f4e591 100644 --- a/cal_date.py +++ b/cal_date.py @@ -130,8 +130,8 @@ def run(start_date, end_date, token): # ======================= if __name__ == "__main__": parser = argparse.ArgumentParser(description="同步交易日历") - parser.add_argument("--start-date", default="20260511", help="开始日期，格式 YYYYMMDD") - parser.add_argument("--end-date", default="20260513", help="结束日期，格式 YYYYMMDD") + parser.add_argument("--start-date", default="20260513", help="开始日期，格式 YYYYMMDD") + parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD") args = parser.parse_args() run(args.start_date, args.end_date, TUSHARE_TOKEN) diff --git a/daily.py b/daily.py index 111bb45..42dac7f 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="20260511", help="开始日期，格式 YYYYMMDD") - parser.add_argument("--end-date", default="20260513", help="结束日期，格式 YYYYMMDD") + parser.add_argument("--start-date", default="20260513", help="开始日期，格式 YYYYMMDD") + parser.add_argument("--end-date", default="20260515", 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 index c1930af..8ff1bc6 100644 --- a/limit_list_d.py +++ b/limit_list_d.py @@ -277,7 +277,7 @@ def run(start_date, end_date): if __name__ == "__main__": - START_DATE = "20260511" # 开始日期，格式 YYYYMMDD 或 YYYY-MM-DD - END_DATE = "20260513" # 结束日期，格式 YYYYMMDD 或 YYYY-MM-DD + START_DATE = "20260513" # 开始日期，格式 YYYYMMDD 或 YYYY-MM-DD + END_DATE = "20260515" # 结束日期，格式 YYYYMMDD 或 YYYY-MM-DD run(START_DATE, END_DATE) diff --git a/limit_up.py b/limit_up.py index bb31051..7467f5a 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-05-11", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") - parser.add_argument("--end-date", default="2026-05-13", help="结束日期，格式 YYYY-MM-DD") + parser.add_argument("--start-date", default="2026-05-13", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期") + parser.add_argument("--end-date", default="2026-05-15", help="结束日期，格式 YYYY-MM-DD") parser.add_argument( "--cleanup-only", action="store_true", diff --git a/run_pipeline.py b/run_pipeline.py index 53e30b4..1cbbb25 100644 --- a/run_pipeline.py +++ b/run_pipeline.py @@ -31,7 +31,10 @@ def main(): today = datetime.now().date() parser = argparse.ArgumentParser( - description="按 basic -> cal_date -> daily -> limit_up -> limit_list_d 顺序同步股票数据" + description=( + "按 basic -> cal_date -> daily -> limit_up -> limit_list_d " + "-> second_board_profit_cache 顺序同步股票数据" + ) ) parser.add_argument( "--date", @@ -64,11 +67,18 @@ def main(): 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 + from second_board_profit_cache import run as run_second_board_profit_cache 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) + run_stage( + "second_board_profit_cache 二板获利缓存", + run_second_board_profit_cache, + start_yyyymmdd, + end_yyyymmdd, + ) print("\n全部流程执行完成") diff --git a/second_board_profit_cache.py b/second_board_profit_cache.py new file mode 100644 index 0000000..ef65f31 --- /dev/null +++ b/second_board_profit_cache.py @@ -0,0 +1,468 @@ +import argparse +from datetime import datetime +from decimal import Decimal, ROUND_HALF_UP + +from sqlalchemy import text + +from stock_config import get_engine + + +DETAIL_TABLE = "second_board_profit_detail_cache" +SUMMARY_TABLE = "second_board_profit_summary_cache" +BATCH_SIZE = 1000 + + +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_decimal(value): + if value is None: + return None + return Decimal(str(value)) + + +def round2(value): + if value is None: + return None + return value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP) + + +def pct_change(value, base): + value = to_decimal(value) + base = to_decimal(base) + if value is None or base is None or base <= 0: + return None + return round2((value - base) / base * Decimal("100")) + + +def format_limit_time(value): + text_value = "" if value is None else str(value).strip() + if not text_value or text_value.lower() in ("none", "null", "nan"): + return "" + if ":" in text_value: + return text_value + if len(text_value) == 4: + text_value = f"{text_value}00" + elif len(text_value) == 5: + text_value = f"0{text_value}" + elif len(text_value) != 6: + return "" + return f"{text_value[0:2]}:{text_value[2:4]}:{text_value[4:6]}" + + +def time_seconds(value): + text_value = format_limit_time(value) + if not text_value: + return None + + parts = text_value.split(":") + if len(parts) < 2 or len(parts) > 3: + return None + + try: + hour = int(parts[0]) + minute = int(parts[1]) + second = int(parts[2]) if len(parts) == 3 else 0 + except ValueError: + return None + + if hour > 23 or minute > 59 or second > 59 or min(hour, minute, second) < 0: + return None + return hour * 3600 + minute * 60 + second + + +def is_target_limit_time(value): + seconds = time_seconds(value) + if seconds is None: + return False + return 9 * 3600 + 31 * 60 <= seconds <= 9 * 3600 + 59 * 60 + 59 + + +def init_db(engine): + detail_sql = f""" + CREATE TABLE IF NOT EXISTS `{DETAIL_TABLE}` ( + `id` BIGINT NOT NULL AUTO_INCREMENT COMMENT '自增主键', + `trade_date` DATE NOT NULL COMMENT 'T日交易日', + `previous_trade_date` DATE NOT NULL COMMENT 'T-1交易日', + `previous_two_trade_date` DATE NOT NULL COMMENT 'T-2交易日', + `next_trade_date` DATE NULL COMMENT 'T+1交易日', + `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码', + `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码', + `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称', + `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块', + `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业', + `previous_close` DECIMAL(16,4) NULL COMMENT 'T-1收盘价', + `limit_price` DECIMAL(16,4) NULL COMMENT 'T日涨停价', + `t_high` DECIMAL(16,4) NULL COMMENT 'T日最高价', + `t_close` DECIMAL(16,4) NULL COMMENT 'T日收盘价', + `t_pct_chg` DECIMAL(10,4) NULL COMMENT 'T日涨跌幅', + `status` VARCHAR(10) NOT NULL DEFAULT '' COMMENT 'limit封板 break炸板', + `first_limit_time` VARCHAR(8) NOT NULL DEFAULT '' COMMENT 'T日首次涨停时间', + `open_times` INT NULL COMMENT '开板次数', + `next_high` DECIMAL(16,4) NULL COMMENT 'T+1最高价', + `next_close` DECIMAL(16,4) NULL COMMENT 'T+1收盘价', + `next_intraday_profitable` TINYINT(1) NULL COMMENT 'T+1盘中是否盈利', + `next_high_profit_pct` DECIMAL(10,2) NULL COMMENT 'T+1最大获利百分比', + `created_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + PRIMARY KEY (`id`), + UNIQUE KEY `uniq_trade_code` (`trade_date`, `ts_code`), + KEY `idx_ts_code_date` (`ts_code`, `trade_date`), + KEY `idx_trade_date` (`trade_date`), + KEY `idx_total_profit` (`next_high_profit_pct`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + """ + summary_sql = f""" + CREATE TABLE IF NOT EXISTS `{SUMMARY_TABLE}` ( + `ts_code` VARCHAR(20) NOT NULL COMMENT 'TuShare股票代码', + `symbol` VARCHAR(10) NOT NULL DEFAULT '' COMMENT '股票代码', + `name` VARCHAR(50) NOT NULL DEFAULT '' COMMENT '股票简称', + `market` VARCHAR(20) NOT NULL DEFAULT '' COMMENT '所属板块', + `industry` VARCHAR(100) NOT NULL DEFAULT '' COMMENT '所属行业', + `second_board_count` INT NOT NULL DEFAULT 0 COMMENT '二板次数', + `limit_count` INT NOT NULL DEFAULT 0 COMMENT '封板次数', + `break_count` INT NOT NULL DEFAULT 0 COMMENT '炸板次数', + `profit_data_count` INT NOT NULL DEFAULT 0 COMMENT '具备收益数据次数', + `total_next_high_profit_pct` DECIMAL(12,2) NOT NULL DEFAULT 0 COMMENT '最大获利累计百分比', + `average_next_high_profit_pct` DECIMAL(10,2) NOT NULL DEFAULT 0 COMMENT '最大获利平均百分比', + `first_trade_date` DATE NULL COMMENT '首次命中日期', + `last_trade_date` DATE NULL COMMENT '最近命中日期', + `updated_at` TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间', + PRIMARY KEY (`ts_code`), + KEY `idx_total_profit` (`total_next_high_profit_pct`), + KEY `idx_last_trade_date` (`last_trade_date`) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + """ + with engine.begin() as conn: + conn.execute(text(detail_sql)) + conn.execute(text(summary_sql)) + print(f"✅ 表 {DETAIL_TABLE} / {SUMMARY_TABLE} 检查完成。") + + +def load_candidate_rows(conn, start_date, end_date): + query = text( + """ + WITH calendar AS ( + SELECT + cal_date AS trade_date, + LAG(cal_date) OVER (ORDER BY cal_date) AS previous_trade_date, + LAG(cal_date, 2) OVER (ORDER BY cal_date) AS previous_two_trade_date, + LEAD(cal_date) OVER (ORDER BY cal_date) AS next_trade_date + FROM trade_calendar + WHERE exchange = 'SSE' AND is_open = 1 + ), + calendar_window AS ( + SELECT * + FROM calendar + WHERE trade_date BETWEEN :start_date AND :end_date + ) + SELECT + DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date, + DATE_FORMAT(previous_trade_date, '%Y-%m-%d') AS previous_trade_date, + DATE_FORMAT(previous_two_trade_date, '%Y-%m-%d') AS previous_two_trade_date, + COALESCE(DATE_FORMAT(next_trade_date, '%Y-%m-%d'), '') AS next_trade_date, + COALESCE(symbol, '') AS symbol, + ts_code, + name, + market, + industry, + previous_close, + limit_price, + t_high, + t_close, + t_pct_chg, + status, + first_limit_time, + open_times, + next_high, + next_close + FROM ( + SELECT + c.trade_date, + c.previous_trade_date, + c.previous_two_trade_date, + c.next_trade_date, + d.symbol, + pl.ts_code, + COALESCE(NULLIF(sb.name, ''), NULLIF(pl.name, ''), '') AS name, + COALESCE(NULLIF(sb.market, ''), '') AS market, + COALESCE(NULLIF(sb.industry, ''), NULLIF(pl.industry, ''), '') AS industry, + COALESCE(d.pre_close, pl.close) AS previous_close, + ROUND( + COALESCE(d.pre_close, pl.close) * + CASE + WHEN COALESCE(sb.market, '') = '北交所' THEN 1.30 + WHEN COALESCE(sb.market, '') IN ('创业板', '科创板') THEN 1.20 + WHEN COALESCE(sb.name, pl.name, '') LIKE '%ST%' THEN 1.05 + ELSE 1.10 + END, + 2 + ) AS limit_price, + d.high AS t_high, + d.close AS t_close, + d.pct_chg AS t_pct_chg, + CASE + WHEN ABS(d.close - d.high) < 0.0001 THEN 'limit' + WHEN d.close < d.high THEN 'break' + ELSE 'unknown' + END AS status, + CASE + WHEN ABS(d.close - d.high) < 0.0001 + THEN NULLIF(NULLIF(TRIM(tlu.first_time), ''), 'None') + WHEN d.close < d.high + THEN NULLIF(NULLIF(TRIM(tlz.first_time), ''), 'None') + ELSE NULL + END AS first_limit_time, + CASE + WHEN ABS(d.close - d.high) < 0.0001 THEN tlu.open_times + WHEN d.close < d.high THEN tlz.open_times + ELSE NULL + END AS open_times, + nd.high AS next_high, + nd.close AS next_close + FROM calendar_window c + INNER JOIN limit_list_d pl + ON pl.trade_date = c.previous_trade_date AND pl.`limit` = 'U' + LEFT JOIN limit_list_d ppl + ON ppl.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + AND ppl.trade_date = c.previous_two_trade_date + AND ppl.`limit` = 'U' + INNER JOIN stock_daily d + ON d.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + AND d.trade_date = c.trade_date + LEFT JOIN stock_daily nd + ON nd.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + AND nd.trade_date = c.next_trade_date + LEFT JOIN limit_list_d tlu + ON tlu.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + AND tlu.trade_date = c.trade_date + AND tlu.`limit` = 'U' + LEFT JOIN limit_list_d tlz + ON tlz.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + AND tlz.trade_date = c.trade_date + AND tlz.`limit` = 'Z' + LEFT JOIN stock_basic sb + ON sb.ts_code COLLATE utf8mb4_unicode_ci = pl.ts_code COLLATE utf8mb4_unicode_ci + WHERE c.previous_trade_date IS NOT NULL + AND c.previous_two_trade_date IS NOT NULL + AND ppl.ts_code IS NULL + AND d.high IS NOT NULL + AND d.close IS NOT NULL + ) candidate + WHERE previous_close IS NOT NULL + AND limit_price IS NOT NULL + AND limit_price > 0 + AND t_high >= limit_price - 0.0001 + AND status IN ('limit', 'break') + ORDER BY trade_date ASC, ts_code ASC + """ + ) + return [ + dict(row) + for row in conn.execute( + query, + {"start_date": start_date, "end_date": end_date}, + ).mappings() + ] + + +def build_detail_records(rows): + records = [] + for row in rows: + first_limit_time = format_limit_time(row["first_limit_time"]) + if not is_target_limit_time(first_limit_time): + continue + + next_intraday_profitable = None + if row["next_high"] is not None and row["t_high"] is not None: + next_intraday_profitable = ( + 1 if to_decimal(row["next_high"]) > to_decimal(row["t_high"]) else 0 + ) + + records.append( + { + "trade_date": row["trade_date"], + "previous_trade_date": row["previous_trade_date"], + "previous_two_trade_date": row["previous_two_trade_date"], + "next_trade_date": row["next_trade_date"] or None, + "symbol": row["symbol"] or "", + "ts_code": row["ts_code"], + "name": row["name"] or "", + "market": row["market"] or "", + "industry": row["industry"] or "", + "previous_close": row["previous_close"], + "limit_price": row["limit_price"], + "t_high": row["t_high"], + "t_close": row["t_close"], + "t_pct_chg": row["t_pct_chg"], + "status": row["status"], + "first_limit_time": first_limit_time, + "open_times": row["open_times"], + "next_high": row["next_high"], + "next_close": row["next_close"], + "next_intraday_profitable": next_intraday_profitable, + "next_high_profit_pct": pct_change(row["next_high"], row["t_high"]), + } + ) + return records + + +def load_existing_keys(conn, start_date, end_date): + query = text( + f""" + SELECT DATE_FORMAT(trade_date, '%Y-%m-%d') AS trade_date, ts_code + FROM `{DETAIL_TABLE}` + WHERE trade_date BETWEEN :start_date AND :end_date + """ + ) + rows = conn.execute( + query, + {"start_date": start_date, "end_date": end_date}, + ).mappings() + return {(row["trade_date"], row["ts_code"]) for row in rows} + + +def write_detail_records(conn, records, start_date, end_date): + existing_keys = load_existing_keys(conn, start_date, end_date) + overwritten_count = sum( + 1 for record in records if (record["trade_date"], record["ts_code"]) in existing_keys + ) + delete_result = conn.execute( + text( + f""" + DELETE FROM `{DETAIL_TABLE}` + WHERE trade_date BETWEEN :start_date AND :end_date + """ + ), + {"start_date": start_date, "end_date": end_date}, + ) + + if not records: + return 0, overwritten_count, delete_result.rowcount + + sql = text( + f""" + INSERT INTO `{DETAIL_TABLE}` ( + trade_date, previous_trade_date, previous_two_trade_date, next_trade_date, + symbol, ts_code, name, market, industry, previous_close, limit_price, + t_high, t_close, t_pct_chg, status, first_limit_time, open_times, + next_high, next_close, next_intraday_profitable, next_high_profit_pct + ) VALUES ( + :trade_date, :previous_trade_date, :previous_two_trade_date, :next_trade_date, + :symbol, :ts_code, :name, :market, :industry, :previous_close, :limit_price, + :t_high, :t_close, :t_pct_chg, :status, :first_limit_time, :open_times, + :next_high, :next_close, :next_intraday_profitable, :next_high_profit_pct + ) + ON DUPLICATE KEY UPDATE + previous_trade_date = VALUES(previous_trade_date), + previous_two_trade_date = VALUES(previous_two_trade_date), + next_trade_date = VALUES(next_trade_date), + symbol = VALUES(symbol), + name = VALUES(name), + market = VALUES(market), + industry = VALUES(industry), + previous_close = VALUES(previous_close), + limit_price = VALUES(limit_price), + t_high = VALUES(t_high), + t_close = VALUES(t_close), + t_pct_chg = VALUES(t_pct_chg), + status = VALUES(status), + first_limit_time = VALUES(first_limit_time), + open_times = VALUES(open_times), + next_high = VALUES(next_high), + next_close = VALUES(next_close), + next_intraday_profitable = VALUES(next_intraday_profitable), + next_high_profit_pct = VALUES(next_high_profit_pct) + """ + ) + for index in range(0, len(records), BATCH_SIZE): + conn.execute(sql, records[index:index + BATCH_SIZE]) + return len(records), overwritten_count, delete_result.rowcount + + +def refresh_summary(conn): + conn.execute(text(f"TRUNCATE TABLE `{SUMMARY_TABLE}`")) + conn.execute( + text( + f""" + INSERT INTO `{SUMMARY_TABLE}` ( + ts_code, symbol, name, market, industry, second_board_count, + limit_count, break_count, profit_data_count, total_next_high_profit_pct, + average_next_high_profit_pct, first_trade_date, last_trade_date + ) + SELECT + ts_code, + MAX(symbol) AS symbol, + MAX(name) AS name, + MAX(market) AS market, + MAX(industry) AS industry, + COUNT(*) AS second_board_count, + SUM(CASE WHEN status = 'limit' THEN 1 ELSE 0 END) AS limit_count, + SUM(CASE WHEN status = 'break' THEN 1 ELSE 0 END) AS break_count, + COUNT(next_high_profit_pct) AS profit_data_count, + ROUND(SUM(COALESCE(next_high_profit_pct, 0)), 2) AS total_next_high_profit_pct, + COALESCE( + ROUND( + SUM(COALESCE(next_high_profit_pct, 0)) + / NULLIF(COUNT(next_high_profit_pct), 0), + 2 + ), + 0 + ) AS average_next_high_profit_pct, + MIN(trade_date) AS first_trade_date, + MAX(trade_date) AS last_trade_date + FROM `{DETAIL_TABLE}` + GROUP BY ts_code + """ + ) + ) + + +def run(start_date, end_date): + start = parse_date(start_date) + end = parse_date(end_date) + if start > end: + raise ValueError("开始日期不能大于结束日期") + + engine = get_engine() + print(f"🚀 开始刷新二板获利缓存：{start} ~ {end}") + + try: + init_db(engine) + with engine.begin() as conn: + rows = load_candidate_rows(conn, start, end) + records = build_detail_records(rows) + write_count, overwritten_count, deleted_count = write_detail_records( + conn, + records, + start, + end, + ) + refresh_summary(conn) + + print(f"候选记录：{len(rows)} 条") + print(f"命中记录：{len(records)} 条") + print(f"清理旧记录：{deleted_count} 条") + print(f"写入记录：{write_count} 条") + print(f"覆盖记录：{overwritten_count} 条") + print("✅ 二板获利缓存刷新完成。") + return True + except Exception as exc: + print(f"❌ 二板获利缓存刷新失败: {exc}") + return False + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="刷新二板获利缓存") + parser.add_argument("--start-date", default="20260515", help="开始日期，格式 YYYYMMDD 或 YYYY-MM-DD") + parser.add_argument("--end-date", default="20260515", help="结束日期，格式 YYYYMMDD 或 YYYY-MM-DD") + args = parser.parse_args() + + run(args.start_date, args.end_date)