Merge remote-tracking branch 'origin/master'

# Conflicts:
#	cal_date.py
#	daily.py
#	limit_up.py

basic.py

@@ -1,9 +1,10 @@
 import tushare as ts
-from sqlalchemy import create_engine
+
+from stock_config import TUSHARE_TOKEN, get_engine
 
 # 1. 配置参数
 # 在 Tushare 官网个人主页获取你的 token
-TOKEN = '0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93'
+TOKEN = TUSHARE_TOKEN
 # 数据库连接配置 (格式: mysql+pymysql://用户名:密码@主机地址:端口/数据库名)
 # DB_CONN_STR = 'mysql+pymysql://yangfan:aA%8023321088@8.159.129.156:10836/level?charset=utf8'
 
@@ -25,18 +26,12 @@ def sync_stock_basic():
 
     if df.empty:
         print("未获取到数据，请检查接口权限或Token设置。")
-        return
+        return False
 
     # 4. 写入数据库
     try:
         # 创建数据库引擎
-        engine = create_engine(
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
+        engine = get_engine()
 
         # 将数据写入数据库表 'stock_basic'
         # if_exists='replace': 如果表存在则替换（适合全量更新基础信息）
@@ -44,10 +39,12 @@ def sync_stock_basic():
         df.to_sql('stock_basic', engine, if_exists='replace', index=False)
 
         print(f"成功将 {len(df)} 条股票基础信息写入数据库表 [stock_basic]。")
+        return True
 
     except Exception as e:
         print(f"写入数据库失败: {e}")
+        return False
 
 
 if __name__ == "__main__":
-    sync_stock_basic()
+    sync_stock_basic()

cal_date.py

@@ -1,7 +1,10 @@
 import tushare as ts
-from sqlalchemy import create_engine, text
+from sqlalchemy import text
 from datetime import datetime
 import sys
+import argparse
+
+from stock_config import TUSHARE_TOKEN, get_engine as create_db_engine
 
 # =======================
 # 1. 日期校验（YYYYMMDD）
@@ -36,14 +39,7 @@ def validate_dates(start_str, end_str):
 # 2. 创建数据库连接（解决 % 编码问题）
 # =======================
 def get_engine():
-    engine = create_engine(
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
-    return engine
+    return create_db_engine()
 
 
 # =======================
@@ -92,11 +88,11 @@ def run(start_date, end_date, token):
         )
     except Exception as e:
         print(f"❌ 获取数据失败: {e}")
-        return
+        return False
 
     if df is None or df.empty:
         print("⚠️ 无数据")
-        return
+        return False
 
     records = []
     for _, row in df.iterrows():
@@ -126,15 +122,16 @@ def run(start_date, end_date, token):
         conn.execute(sql, records)
 
     print(f"✅ 写入完成，共 {len(records)} 条")
+    return True
 
 
 # =======================
 # 5. 程序入口（只改这里）
 # =======================
 if __name__ == "__main__":
-    START_DATE = "20240427"
-    END_DATE   = "20260427"
+    parser = argparse.ArgumentParser(description="同步交易日历")
+    parser.add_argument("--start-date", default="20260429", help="开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260429", help="结束日期，格式 YYYYMMDD")
+    args = parser.parse_args()
 
-    TUSHARE_TOKEN = "0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93"
-
-    run(START_DATE, END_DATE, TUSHARE_TOKEN)
+    run(args.start_date, args.end_date, TUSHARE_TOKEN)

daily.py

@@ -2,29 +2,29 @@ import tushare as ts
 import pymysql
 from datetime import datetime
 import time
+import argparse
 
-# ===== 配置 =====
-TUSHARE_TOKEN = "0fad3cf498757089e2630028455d5bbe1637475788bcdaa1f2175e93"
-
-DB_CONFIG = {
-    "host": "8.159.129.156",
-    "port": 10836,
-    "user": "yangfan",
-    "password": "aA%8023321088",
-    "database": "level",
-    "charset": "utf8mb4"
-}
+from stock_config import DB_CONFIG, TUSHARE_TOKEN
 
 # ===== 初始化 =====
 ts.set_token(TUSHARE_TOKEN)
 pro = ts.pro_api()
 
-conn = pymysql.connect(**DB_CONFIG)
-cursor = conn.cursor()
+conn = None
+cursor = None
+
+
+def get_connection():
+    global conn, cursor
+    if conn is None or not conn.open:
+        conn = pymysql.connect(**DB_CONFIG)
+        cursor = conn.cursor()
+    return conn, cursor
 
 
 # ===== 建表 =====
 def create_tables():
+    db_conn, db_cursor = get_connection()
     sql = """
     CREATE TABLE IF NOT EXISTS stock_daily (
         id BIGINT AUTO_INCREMENT PRIMARY KEY,
@@ -43,8 +43,8 @@ def create_tables():
         UNIQUE KEY uniq_code_date (ts_code, trade_date)
     );
     """
-    cursor.execute(sql)
-    conn.commit()
+    db_cursor.execute(sql)
+    db_conn.commit()
     print("表检查/创建完成")
 
 
@@ -55,6 +55,7 @@ def str_to_date(date_str):
 
 # ===== 获取交易日 =====
 def get_trade_dates(start_date, end_date):
+    _, db_cursor = get_connection()
     sql = """
     SELECT cal_date 
     FROM trade_calendar
@@ -62,14 +63,15 @@ def get_trade_dates(start_date, end_date):
       AND is_open = 1
     ORDER BY cal_date
     """
-    cursor.execute(sql, (str_to_date(start_date), str_to_date(end_date)))
-    results = cursor.fetchall()
+    db_cursor.execute(sql, (str_to_date(start_date), str_to_date(end_date)))
+    results = db_cursor.fetchall()
 
     return [row[0].strftime("%Y%m%d") for row in results]
 
 
 # ===== 插入数据 =====
 def insert_data(df):
+    db_conn, db_cursor = get_connection()
     if df is None or df.empty:
         return 0, 0
 
@@ -99,10 +101,10 @@ def insert_data(df):
             row['amount']
         ))
 
-    cursor.executemany(sql, data)
-    conn.commit()
+    db_cursor.executemany(sql, data)
+    db_conn.commit()
 
-    inserted = cursor.rowcount
+    inserted = db_cursor.rowcount
     skipped = len(data) - inserted
 
     return inserted, skipped
@@ -140,11 +142,14 @@ def run(start_date, end_date):
     print("\n====== 汇总 ======")
     print(f"总新增: {total_inserted}")
     print(f"总跳过: {total_skipped}")
+    return True
 
 
 # ===== 输入参数 =====
 if __name__ == "__main__":
-    START_DATE = "20260427"
-    END_DATE = "20260427"
+    parser = argparse.ArgumentParser(description="同步日行情")
+    parser.add_argument("--start-date", default="20260429", help="开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", default="20260429", help="结束日期，格式 YYYYMMDD")
+    args = parser.parse_args()
 
-    run(START_DATE, END_DATE)
+    run(args.start_date, args.end_date)

limit_up.py

@@ -1,11 +1,19 @@
 import pandas as pd
-import numpy as np
-from sqlalchemy import create_engine, text
+from sqlalchemy import text
+import argparse
+
+from stock_config import get_engine
+
+# --- 数据库配置 ---
+# 请根据实际情况修改数据库连接信息
+# DB_CONN_STR = 'mysql+pymysql://root:password@127.0.0.1:3306/stock_data'
 
 
 def init_db(engine):
     """
     初始化数据库表结构
+    1. 增加 symbol 列，位于 trade_date 之后
+    2. 建立 UNIQUE KEY 防止物理重复写入
     """
     create_table_sql = """
     CREATE TABLE IF NOT EXISTS `stock_limit_data` (
@@ -29,70 +37,42 @@ def analyze_and_save_limits(start_date, end_date):
     """
     计算并在数据库中落地涨停、炸板详情
     """
-    # 数据库连接配置
-    engine = create_engine(
-        "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-        connect_args={
-            "password": "aA%8023321088"
-        },
-        pool_pre_ping=True
-    )
+    engine = get_engine()
 
+    # 0. 初始化检查表结构
     init_db(engine)
 
-    # 1. 获取行情数据
+    # 1. 获取基础数据和行情数据
     query = f"""
     SELECT 
         d.trade_date, d.symbol, d.ts_code, b.name, 
         d.high, d.close, d.pre_close, b.market
     FROM stock_daily d
     JOIN stock_basic b ON d.ts_code = b.ts_code
-    WHERE d.trade_date >= '{start_date}' 
+    WHERE d.trade_date > '{start_date}' 
       AND d.trade_date <= '{end_date}'
-      AND d.pre_close IS NOT NULL
     ORDER BY d.trade_date ASC
     """
 
     try:
         df = pd.read_sql(query, engine)
         if df.empty:
-            print(f"⚠️ 区间 [{start_date} , {end_date}] 内未获取到数据。")
-            return
+            print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。")
+            return True
 
-        # 强制类型转换
-        df['pre_close'] = pd.to_numeric(df['pre_close'], errors='coerce')
-        df['high'] = pd.to_numeric(df['high'], errors='coerce')
-        df['close'] = pd.to_numeric(df['close'], errors='coerce')
-
-        # 2. 涨停价计算规则 (已取消 ST 判定)
+        # 2. 涨停价计算规则
         def calc_limit(row):
-            m = str(row['market'])
-            p = row['pre_close']
-
-            if pd.isna(p) or p <= 0: return 0.0
-
-            # 仅保留板块判定
-            if '创业板' in m or '科创板' in m:
-                ratio = 1.2
-            elif '北交所' in m:
-                ratio = 1.3
-            else:
-                ratio = 1.1  # 普通主板（含ST股现在也按10%计算）
-
-            # 采用标准四舍五入
+            m, p = str(row['market']), float(row['pre_close'])
+            # 主板10%，创业板/科创板20%，北交所30%
+            ratio = 1.2 if ('创业板' in m or '科创板' in m) else (1.3 if '北交所' in m else 1.1)
             return round(p * ratio, 2)
 
         df['limit_price'] = df.apply(calc_limit, axis=1)
 
-        # 3. 筛选逻辑：利用 abs 差值判断，解决浮点数精度问题
-        # 只要最高价触及理论涨停价（误差0.01以内）即视为触板
-        is_at_limit = (df['high'] - df['limit_price']).abs() < 0.011
-        is_close_at_limit = (df['close'] - df['limit_price']).abs() < 0.011
+        # 3. 筛选涨停(1)和炸板(2)
+        df['is_limit'] = (df['high'] == df['limit_price']) & (df['close'] == df['limit_price'])
+        df['is_broken'] = (df['high'] == df['limit_price']) & (df['close'] < df['limit_price'])
 
-        df['is_limit'] = is_at_limit & is_close_at_limit
-        df['is_broken'] = is_at_limit & (~is_close_at_limit)
-
-        # 分离结果
         limit_df = df[df['is_limit']].copy()
         limit_df['status'] = 1
 
@@ -102,33 +82,49 @@ def analyze_and_save_limits(start_date, end_date):
         save_df = pd.concat([limit_df, broken_df])
 
         if not save_df.empty:
+            # 4. 字段清洗与对齐
             final_df = save_df[[
                 'trade_date', 'symbol', 'ts_code', 'name',
                 'status', 'limit_price', 'close', 'market'
             ]]
 
-            # 4. 事务入库：先删后插
+            # 5. 【核心防重逻辑】：开启事务，先删后插
             with engine.begin() as conn:
+                # 获取待写入的所有交易日期
                 target_dates = final_df['trade_date'].astype(str).unique()
                 date_list = ", ".join([f"'{d}'" for d in target_dates])
 
+                # 第一层逻辑防重：删除数据库中已存在的相同日期数据
                 delete_query = f"DELETE FROM stock_limit_data WHERE trade_date IN ({date_list})"
                 conn.execute(text(delete_query))
 
+                # 第二层物理防重：通过 to_sql 写入，若触发 Unique Key 冲突会抛出异常回滚
                 final_df.to_sql('stock_limit_data', conn, if_exists='append', index=False)
 
             print(f"✅ 处理成功！涉及交易日: {len(target_dates)} 天")
-            print(f"📊 记录详情: 总计 {len(final_df)} 条 (涨停: {len(limit_df)} / 炸板: {len(broken_df)})")
+            print(f"📊 总计入库: {len(final_df)} 条记录（已自动去重并覆盖历史数据）。")
         else:
             print("ℹ️ 该时段内全市场无涨停或炸板记录。")
+        return True
 
     except Exception as e:
         print(f"❌ 运行失败: {e}")
+        return False
 
 
+# ============================================================
+# 用户配置区
+# ============================================================
 if __name__ == "__main__":
-    # 配置日期区间
-    START_DATE = '2024-09-23'
-    END_DATE = '2026-04-27'
+    parser = argparse.ArgumentParser(description="计算并保存涨停/炸板数据")
+    parser.add_argument("--start-date", default="2026-04-29", help="开始日期，格式 YYYY-MM-DD，实际查询为大于该日期")
+    parser.add_argument("--end-date", default="2026-04-29", help="结束日期，格式 YYYY-MM-DD")
+    args = parser.parse_args()
 
-    analyze_and_save_limits(START_DATE, END_DATE)
+    # 开始日期：计算从该日期的下一个交易日开始
+    START_DATE = args.start_date
+
+    # 结束日期
+    END_DATE = args.end_date
+
+    analyze_and_save_limits(START_DATE, END_DATE)

luoji.py

@@ -1,106 +0,0 @@
-import pandas as pd
-from sqlalchemy import create_engine
-
-
-def analyze_stock_performance(start_date, end_date):
-    # --- 数据库配置 ---
-    try:
-        engine = create_engine(
-            "mysql+pymysql://yangfan@8.159.129.156:10836/level?charset=utf8mb4",
-            connect_args={
-                "password": "aA%8023321088"
-            },
-            pool_pre_ping=True
-        )
-
-        # 1. 获取涨停当日股票及价格
-        limit_query = f"""
-        SELECT l.ts_code, l.name, d.open AS limit_day_open, d.high AS limit_day_high
-        FROM stock_limit_data l
-        LEFT JOIN stock_daily d ON l.ts_code = d.ts_code AND l.trade_date = d.trade_date
-        WHERE l.trade_date = '{start_date}' AND l.status = 1
-        """
-        df_limit = pd.read_sql(limit_query, engine)
-
-        if df_limit.empty:
-            print(f"提示：{start_date} 当日没有涨停记录。")
-            return
-
-        # 2. 获取行情区间数据
-        codes = df_limit['ts_code'].unique().tolist()
-        codes_tuple = f"('{codes[0]}')" if len(codes) == 1 else str(tuple(codes))
-        daily_query = f"""
-        SELECT ts_code, trade_date, high, low, close
-        FROM stock_daily
-        WHERE ts_code IN {codes_tuple} AND trade_date BETWEEN '{start_date}' AND '{end_date}'
-        ORDER BY trade_date ASC
-        """
-        df_daily = pd.read_sql(daily_query, engine)
-        df_daily['trade_date'] = pd.to_datetime(df_daily['trade_date'])
-
-        # 3. 核心逻辑处理
-        final_report = []
-        for _, limit_row in df_limit.iterrows():
-            code = limit_row['ts_code']
-            stock_data = df_daily[df_daily['ts_code'] == code].copy()
-
-            if stock_data.empty: continue
-
-            # A. 找到区间内的最高价及其第一次出现的日期
-            max_high = float(stock_data['high'].max())
-            # 获取最高价所在的那一行（如果有多个相同最高价，取第一个）
-            max_high_row = stock_data[stock_data['high'] == max_high].iloc[0]
-            max_date = max_high_row['trade_date']
-
-            # B. 寻找最高价日期之后的最低价
-            post_max_data = stock_data[stock_data['trade_date'] > max_date]
-
-            if not post_max_data.empty:
-                min_low = float(post_max_data['low'].min())
-                last_close = float(post_max_data.iloc[-1]['close'])
-
-                price_diff = max_high - min_low
-                drop_pct = (price_diff / max_high * 100)
-                rebound_pct = ((last_close - min_low) / min_low * 100) if min_low > 0 else 0
-            else:
-                # 如果最高点出现在结束日当天，则没有“后期最低价”
-                min_low = None
-                price_diff = 0
-                drop_pct = 0
-                rebound_pct = 0
-
-            final_report.append({
-                "代码": code,
-                "名称": limit_row['name'],
-                "当日开盘": float(limit_row['limit_day_open'] or 0),
-                "当日最高": float(limit_row['limit_day_high'] or 0),
-                "区间最高": max_high,
-                "最高日期": max_date.strftime('%Y-%m-%d'),
-                "见顶后最低": min_low,
-                "最大回落幅%": round(drop_pct, 2),
-                "低位至期末波动%": round(rebound_pct, 2)
-            })
-
-        # 4. 排序与输出
-        if final_report:
-            result_df = pd.DataFrame(final_report)
-            result_df = result_df.sort_values(by="最大回落幅%", ascending=False)
-
-            print(f"\n分析区间: {start_date} 至 {end_date}")
-            print("说明: '见顶后最低' 指最高价日期之后的最低点")
-            print("=" * 120)
-            print(result_df.to_string(index=False, na_rep='N/A'))
-            print("=" * 120)
-
-    except Exception as e:
-        print(f"发生错误: {e}")
-
-
-# ==========================================
-# 设定日期
-# ==========================================
-if __name__ == "__main__":
-    START_INPUT = '2024-09-24'
-    END_INPUT = '2026-04-23'
-
-    analyze_stock_performance(START_INPUT, END_INPUT)

run_pipeline.py

@@ -0,0 +1,75 @@
+import argparse
+from datetime import datetime, timedelta
+
+from stock_config import TUSHARE_TOKEN
+
+
+def parse_yyyymmdd(value):
+    try:
+        return datetime.strptime(value, "%Y%m%d").date()
+    except ValueError as exc:
+        raise argparse.ArgumentTypeError("日期格式应为 YYYYMMDD") from exc
+
+
+def to_yyyymmdd(date_value):
+    return date_value.strftime("%Y%m%d")
+
+
+def to_iso_date(date_value):
+    return date_value.strftime("%Y-%m-%d")
+
+
+def run_stage(name, func, *args):
+    print(f"\n========== {name} 开始 ==========")
+    result = func(*args)
+    if result is False:
+        raise RuntimeError(f"{name} 执行失败，流程已停止")
+    print(f"========== {name} 完成 ==========")
+
+
+def main():
+    today = datetime.now().date()
+
+    parser = argparse.ArgumentParser(
+        description="按 basic -> cal_date -> daily -> limit_up 顺序同步股票数据"
+    )
+    parser.add_argument(
+        "--date",
+        type=parse_yyyymmdd,
+        default=today,
+        help="同步单日数据，格式 YYYYMMDD；未指定时默认今天",
+    )
+    parser.add_argument("--start-date", type=parse_yyyymmdd, help="区间开始日期，格式 YYYYMMDD")
+    parser.add_argument("--end-date", type=parse_yyyymmdd, help="区间结束日期，格式 YYYYMMDD")
+    parser.add_argument("--skip-basic", action="store_true", help="跳过股票基础信息同步")
+    args = parser.parse_args()
+
+    start_date = args.start_date or args.date
+    end_date = args.end_date or args.date
+
+    if start_date > end_date:
+        raise SystemExit("开始日期不能大于结束日期")
+
+    start_yyyymmdd = to_yyyymmdd(start_date)
+    end_yyyymmdd = to_yyyymmdd(end_date)
+    limit_start = to_iso_date(start_date - timedelta(days=1))
+    limit_end = to_iso_date(end_date)
+
+    if not args.skip_basic:
+        from basic import sync_stock_basic
+
+        run_stage("basic 股票基础信息", sync_stock_basic)
+
+    from cal_date import run as run_cal_date
+    from daily import run as run_daily
+    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)
+
+    print("\n全部流程执行完成")
+
+
+if __name__ == "__main__":
+    main()

stock_config.py

@@ -0,0 +1,33 @@
+import os
+
+
+TUSHARE_TOKEN = os.getenv(
+    "TUSHARE_TOKEN",
+    "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"),
+}
+
+
+def get_db_url():
+    return (
+        f"mysql+pymysql://{DB_CONFIG['user']}@{DB_CONFIG['host']}:"
+        f"{DB_CONFIG['port']}/{DB_CONFIG['database']}?charset={DB_CONFIG['charset']}"
+    )
+
+
+def get_engine():
+    from sqlalchemy import create_engine
+
+    return create_engine(
+        get_db_url(),
+        connect_args={"password": DB_CONFIG["password"]},
+        pool_pre_ping=True,
+    )