From 1b39ad68871a17d70fa3e1ce5dee7cceded3805e Mon Sep 17 00:00:00 2001 From: fish Date: Sat, 18 Jul 2026 16:25:52 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E9=87=8F=E6=9B=B4=E6=96=B0=E9=80=BB?= =?UTF-8?q?=E8=BE=91=EF=BC=8C=E6=94=AF=E6=8C=81=E4=BA=A4=E6=98=93=E6=97=A5?= =?UTF-8?q?=E5=88=A4=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.7 --- update.py | 265 ++++++++++++++++++++++++++++++++++++++---------------- 1 file changed, 190 insertions(+), 75 deletions(-) diff --git a/update.py b/update.py index 3598e6c..d150b6c 100644 --- a/update.py +++ b/update.py @@ -1,6 +1,7 @@ import csv import os import sqlite3 +import sys import time from datetime import datetime @@ -19,6 +20,20 @@ FUT_DAILY_FIELDS = [ ] +def is_trading_day(exchange="CZCE"): + """检查指定交易所今天是否开市""" + today = datetime.now().strftime("%Y%m%d") + rows = tushare_query( + "trade_cal", + params={"exchange": exchange, "start_date": today, "end_date": today}, + fields="cal_date,is_open", + ) + if rows: + return rows[0]["is_open"] == "1" + # 查不到（未来日期未更新）时保守放行 + return True + + def tushare_query(api_name, params=None, fields=None): req = {"api_name": api_name, "token": TOKEN} if params: @@ -37,8 +52,22 @@ def tushare_query(api_name, params=None, fields=None): return [dict(zip(fields_list, item)) for item in items] +def is_trading_day(exchange="CZCE"): + """检查指定交易所今天是否开市""" + today = datetime.now().strftime("%Y%m%d") + rows = tushare_query( + "trade_cal", + params={"exchange": exchange, "start_date": today, "end_date": today}, + fields="cal_date,is_open", + ) + if rows: + return rows[0]["is_open"] == "1" + return True # 查不到时保守放行 + + +# ── 合约列表 ────────────────────────────────── + def get_active_contracts(exchange, fut_code): - """获取指定交易所、品种仍在交易的合约列表""" contracts = tushare_query( "fut_basic", params={"exchange": exchange, "fut_code": fut_code, "fut_type": "1"}, @@ -49,7 +78,6 @@ def get_active_contracts(exchange, fut_code): def get_all_contracts(exchange, fut_code): - """获取指定交易所、品种的全部合约（含已退市）""" return tushare_query( "fut_basic", params={"exchange": exchange, "fut_code": fut_code, "fut_type": "1"}, @@ -57,14 +85,90 @@ def get_all_contracts(exchange, fut_code): ) -def save_contract_csv(fut_code, contracts): - """对每个活跃合约拉取日线行情并写入 data/{fut_code}/{code}.csv""" +# ── 增量更新 CSV ────────────────────────────── + +def update_contract_csv(fut_code, contract): + """ + 增量更新单份合约 CSV。 + 从已有数据的最后交易日重新拉取（含该日），实现补漏 + 修正。 + 返回新增/修改的行，或 None（无变化）。 + """ + ts_code = contract["ts_code"] + csv_name = ts_code.split(".")[0] + ".csv" + out_dir = os.path.join(DATA_DIR, fut_code) + csv_path = os.path.join(out_dir, csv_name) + os.makedirs(out_dir, exist_ok=True) + + # 读取已有数据，获取最后交易日 + existing = [] + if os.path.exists(csv_path): + with open(csv_path) as f: + reader = csv.DictReader(f) + existing = list(reader) + + last_date = existing[-1]["trade_date"] if existing else None + + # 从最后交易日开始拉取（包含该日，以便修正） + params = {"ts_code": ts_code} + if last_date: + params["start_date"] = last_date + + rows = tushare_query("fut_daily", params=params, fields=",".join(FUT_DAILY_FIELDS)) + if not rows: + return None + + rows.sort(key=lambda r: r["trade_date"]) + + if not existing: + # 全新合约 + with open(csv_path, "w", newline="") as f: + w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) + w.writeheader() + w.writerows(rows) + return rows + + if rows[0]["trade_date"] == last_date: + # 最后交易日可能有修正 → 替换最后一行再追加 + existing = existing[:-1] + merged = existing + rows + elif rows[0]["trade_date"] > last_date: + # 纯新增数据 + merged = existing + rows + else: + merged = existing + rows # 安全兜底 + + with open(csv_path, "w", newline="") as f: + w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) + w.writeheader() + w.writerows(merged) + + return rows + + +def update_all_contracts(fut_code, contracts): + """批量增量更新，返回所有有变化的行""" + all_new = [] + for c in contracts: + code = c["ts_code"].split(".")[0] + new_rows = update_contract_csv(fut_code, c) + if new_rows is None: + print(f" {code:12s} 无新数据") + else: + print(f" {code:12s} +{len(new_rows)} 条 ({new_rows[0]['trade_date']} ~ {new_rows[-1]['trade_date']})") + all_new.extend(new_rows) + time.sleep(0.3) + return all_new + + +# ── 全量拉取（初始化用） ───────────────────── + +def fetch_all_csv(fut_code, contracts): out_dir = os.path.join(DATA_DIR, fut_code) os.makedirs(out_dir, exist_ok=True) for c in contracts: - ts_code = c["ts_code"] # e.g. FG2609.ZCE - csv_name = ts_code.split(".")[0] + ".csv" # FG2609.csv + ts_code = c["ts_code"] + csv_name = ts_code.split(".")[0] + ".csv" csv_path = os.path.join(out_dir, csv_name) rows = tushare_query( @@ -73,20 +177,20 @@ def save_contract_csv(fut_code, contracts): fields=",".join(FUT_DAILY_FIELDS), ) if not rows: - print(f" {csv_name:12s} 无数据，跳过") + print(f" {csv_name:12s} 无数据，跳过") continue rows.sort(key=lambda r: r["trade_date"]) - with open(csv_path, "w", newline="") as f: w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) w.writeheader() w.writerows(rows) - print(f" {csv_name:12s} {len(rows)} 条记录 ({rows[0]['trade_date']} ~ {rows[-1]['trade_date']})") + print(f" {csv_name:12s} {len(rows)} 条 ({rows[0]['trade_date']} ~ {rows[-1]['trade_date']})") + time.sleep(0.3) - time.sleep(0.3) # tushare 频率限制 +# ── 数据库 ──────────────────────────────────── def init_db(): conn = sqlite3.connect(DB_PATH) @@ -94,32 +198,19 @@ def init_db(): CREATE TABLE IF NOT EXISTS daily ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, - pre_close REAL, - pre_settle REAL, - open REAL, - high REAL, - low REAL, - close REAL, - settle REAL, - change1 REAL, - change2 REAL, - vol REAL, - amount REAL, - oi REAL, - oi_chg REAL, + pre_close REAL, pre_settle REAL, + open REAL, high REAL, low REAL, close REAL, settle REAL, + change1 REAL, change2 REAL, + vol REAL, amount REAL, oi REAL, oi_chg REAL, delv_settle REAL, PRIMARY KEY (ts_code, trade_date) ) """) conn.execute(""" CREATE TABLE IF NOT EXISTS contracts ( - ts_code TEXT PRIMARY KEY, - symbol TEXT, - name TEXT, - exchange TEXT, - fut_code TEXT, - list_date TEXT, - delist_date TEXT + ts_code TEXT PRIMARY KEY, + symbol TEXT, name TEXT, exchange TEXT, + fut_code TEXT, list_date TEXT, delist_date TEXT ) """) conn.commit() @@ -131,78 +222,102 @@ def sync_contracts_to_db(fut_code, exchange, contracts): cursor = conn.cursor() for c in contracts: cursor.execute( - "INSERT OR REPLACE INTO contracts (ts_code, symbol, name, exchange, fut_code, list_date, delist_date) " - "VALUES (?, ?, ?, ?, ?, ?, ?)", - (c["ts_code"], c["symbol"], c["name"], exchange, fut_code, c["list_date"], c.get("delist_date")), + "INSERT OR REPLACE INTO contracts VALUES (?, ?, ?, ?, ?, ?, ?)", + (c["ts_code"], c["symbol"], c["name"], exchange, fut_code, + c["list_date"], c.get("delist_date")), ) conn.commit() conn.close() - print(f" contracts 表写入 {len(contracts)} 条") + print(f" contracts 表同步 {len(contracts)} 条") -def sync_csv_to_db(): +def sync_rows_to_db(rows): + """只同步有变化的行到 daily 表（增量更新用）""" + if not rows: + return + conn = sqlite3.connect(DB_PATH) + placeholders = ",".join("?" for _ in FUT_DAILY_FIELDS) + cols = ",".join(FUT_DAILY_FIELDS) + sql = f"INSERT OR REPLACE INTO daily ({cols}) VALUES ({placeholders})" + vals = [[r.get(c) for c in FUT_DAILY_FIELDS] for r in rows] + conn.executemany(sql, vals) + conn.commit() + conn.close() + print(f" daily 表更新 {len(rows)} 条") + + +def sync_all_csv_to_db(): + """全量同步所有 CSV 到 daily 表（初始化用）""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() - for root, _, files in os.walk(DATA_DIR): for fname in sorted(files): if not fname.endswith(".csv"): continue - fpath = os.path.join(root, fname) with open(fpath) as f: reader = csv.DictReader(f) rows = list(reader) - if not rows: continue - placeholders = ",".join("?" for _ in FUT_DAILY_FIELDS) cols = ",".join(FUT_DAILY_FIELDS) sql = f"INSERT OR REPLACE INTO daily ({cols}) VALUES ({placeholders})" vals = [[r.get(c) for c in FUT_DAILY_FIELDS] for r in rows] cursor.executemany(sql, vals) - - print(f" {fname:12s} {len(rows)} 条 -> daily 表") - + print(f" {fname:12s} {len(rows)} 条") conn.commit() conn.close() +# ── 主入口 ──────────────────────────────────── + +FG_EXTRA_CODES = {f"FG{suffix}.ZCE" for suffix in + [f"14{i:02d}" for i in range(1, 13)] + + [f"15{i:02d}" for i in range(1, 13)] + + [f"16{i:02d}" for i in range(1, 13)] + + [f"17{i:02d}" for i in range(1, 13)] + + [f"18{i:02d}" for i in range(1, 13)] + + [f"19{i:02d}" for i in range(1, 13)] + + [f"20{i:02d}" for i in range(1, 13)] + + [f"21{i:02d}" for i in range(1, 13)] + + [f"22{i:02d}" for i in range(1, 13)] + + [f"23{i:02d}" for i in range(1, 13)] + + [f"24{i:02d}" for i in range(1, 13)] + + [f"25{i:02d}" for i in range(1, 13)] + + [f"26{i:02d}" for i in range(1, 8)]} + + if __name__ == "__main__": - # 活跃合约 - active = get_active_contracts("CZCE", "FG") - # 已退市合约：FG1401-FG1412 ~ FG2601-FG2607 - extra_codes = {f"FG{suffix}.ZCE" for suffix in - [f"14{i:02d}" for i in range(1, 13)] + - [f"15{i:02d}" for i in range(1, 13)] + - [f"16{i:02d}" for i in range(1, 13)] + - [f"17{i:02d}" for i in range(1, 13)] + - [f"18{i:02d}" for i in range(1, 13)] + - [f"19{i:02d}" for i in range(1, 13)] + - [f"20{i:02d}" for i in range(1, 13)] + - [f"21{i:02d}" for i in range(1, 13)] + - [f"22{i:02d}" for i in range(1, 13)] + - [f"23{i:02d}" for i in range(1, 13)] + - [f"24{i:02d}" for i in range(1, 13)] + - [f"25{i:02d}" for i in range(1, 13)] + - [f"26{i:02d}" for i in range(1, 8)]} - all_fg = get_all_contracts("CZCE", "FG") - active_codes = {c["ts_code"] for c in active} - extra = [c for c in all_fg if c["ts_code"] in extra_codes and c["ts_code"] not in active_codes] - contracts = active + extra + full_init = "--init" in sys.argv - print(f"玻璃合约共 {len(contracts)} 个（含 {len(extra)} 个已退市）:") - - for c in contracts: - code = c["ts_code"].split(".")[0] - print(f" {code:8s} 上市:{c['list_date']} 退市:{c.get('delist_date','-')}") - - print("\n拉取日线行情...") - save_contract_csv("FG", contracts) - - print("\n写入数据库...") init_db() - sync_contracts_to_db("FG", "CZCE", contracts) - sync_csv_to_db() + active = get_active_contracts("CZCE", "FG") + + if full_init: + all_fg = get_all_contracts("CZCE", "FG") + active_codes = {c["ts_code"] for c in active} + extra = [c for c in all_fg if c["ts_code"] in FG_EXTRA_CODES and c["ts_code"] not in active_codes] + contracts = active + extra + print(f"全量初始化: {len(contracts)} 个合约") + for c in contracts: + code = c["ts_code"].split(".")[0] + print(f" {code:8s} 上市:{c['list_date']} 退市:{c.get('delist_date','-')}") + fetch_all_csv("FG", contracts) + sync_contracts_to_db("FG", "CZCE", contracts) + sync_all_csv_to_db() + else: + if not is_trading_day(): + print("今天非交易日，无需更新") + print("完成") + sys.exit(0) + + print(f"增量更新: {len(active)} 个活跃合约") + new_rows = update_all_contracts("FG", active) + if new_rows: + sync_contracts_to_db("FG", "CZCE", active) + sync_rows_to_db(new_rows) + else: + print(" 无新数据") + print("完成")