From e1aa6ab1795cf1a4eae2fde5c20f17fc6e0f090b Mon Sep 17 00:00:00 2001 From: yangfan <784310601@qq.com> Date: Wed, 29 Apr 2026 17:10:21 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E4=B8=80=E9=94=AE=E8=BF=90?= =?UTF-8?q?=E8=A1=8C=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- basic.py | 19 ++++----- cal_date.py | 29 ++++++------- daily.py | 51 ++++++++++++----------- limit_up.py | 28 +++++++------ luoji.py | 106 ------------------------------------------------ run_pipeline.py | 75 ++++++++++++++++++++++++++++++++++ stock_config.py | 33 +++++++++++++++ 7 files changed, 173 insertions(+), 168 deletions(-) delete mode 100644 luoji.py create mode 100644 run_pipeline.py create mode 100644 stock_config.py diff --git a/basic.py b/basic.py index 546f06a..2041a4f 100644 --- a/basic.py +++ b/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() \ No newline at end of file + sync_stock_basic() diff --git a/cal_date.py b/cal_date.py index 9317a88..546171a 100644 --- a/cal_date.py +++ b/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 = "20240424" - END_DATE = "20260424" + 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) \ No newline at end of file + run(args.start_date, args.end_date, TUSHARE_TOKEN) diff --git a/daily.py b/daily.py index feb2a3f..86f5015 100644 --- a/daily.py +++ b/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 = "20260424" - END_DATE = "20260424" + 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) \ No newline at end of file + run(args.start_date, args.end_date) diff --git a/limit_up.py b/limit_up.py index c36fbda..6174c54 100644 --- a/limit_up.py +++ b/limit_up.py @@ -1,5 +1,8 @@ import pandas as pd -from sqlalchemy import create_engine, text +from sqlalchemy import text +import argparse + +from stock_config import get_engine # --- 数据库配置 --- # 请根据实际情况修改数据库连接信息 @@ -34,13 +37,7 @@ 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) @@ -61,7 +58,7 @@ def analyze_and_save_limits(start_date, end_date): df = pd.read_sql(query, engine) if df.empty: print(f"⚠️ 在指定区间 ({start_date} -> {end_date}] 内未获取到数据。") - return + return True # 2. 涨停价计算规则 def calc_limit(row): @@ -108,19 +105,26 @@ def analyze_and_save_limits(start_date, end_date): print(f"📊 总计入库: {len(final_df)} 条记录（已自动去重并覆盖历史数据）。") else: print("ℹ️ 该时段内全市场无涨停或炸板记录。") + 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("--end-date", default="2026-04-29", help="结束日期，格式 YYYY-MM-DD") + args = parser.parse_args() + # 开始日期：计算从该日期的下一个交易日开始 - START_DATE = '2026-04-23' + START_DATE = args.start_date # 结束日期 - END_DATE = '2026-04-24' + END_DATE = args.end_date - analyze_and_save_limits(START_DATE, END_DATE) \ No newline at end of file + analyze_and_save_limits(START_DATE, END_DATE) diff --git a/luoji.py b/luoji.py deleted file mode 100644 index 248cbf6..0000000 --- a/luoji.py +++ /dev/null @@ -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) \ No newline at end of file diff --git a/run_pipeline.py b/run_pipeline.py new file mode 100644 index 0000000..48dc85f --- /dev/null +++ b/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() diff --git a/stock_config.py b/stock_config.py new file mode 100644 index 0000000..1790a6f --- /dev/null +++ b/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, + )