yang/python
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增一键运行功能

Browse Source
This commit is contained in:
yangfan
2026-04-29 17:10:21 +08:00
parent f843197e27
commit e1aa6ab179
7 changed files with 173 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files
basic.py
+7 -10
View File
@@ -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,9 +39,11 @@ 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__":
cal_date.py
+13 -16
View File
@@ -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)
run(args.start_date, args.end_date, TUSHARE_TOKEN)
daily.py
+28 -23
View File
@@ -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)
run(args.start_date, args.end_date)
limit_up.py
+15 -11
View File
@@ -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)
luoji.py
-106
View File
@@ -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
+75
View File
@@ -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
+33
View File
@@ -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,
)