import csv import os import sqlite3 import time from datetime import datetime import requests TOKEN = "76efd8465f9f2591aa42a385268e06acf6b80b7a15be2267ad2281b7" API_URL = "https://api.tushare.pro" DATA_DIR = os.path.join(os.path.dirname(__file__), "data") DB_PATH = os.path.join(os.path.dirname(__file__), "db", "futures.db") FUT_DAILY_FIELDS = [ "ts_code", "trade_date", "pre_close", "pre_settle", "open", "high", "low", "close", "settle", "change1", "change2", "vol", "amount", "oi", "oi_chg", "delv_settle", ] def tushare_query(api_name, params=None, fields=None): req = {"api_name": api_name, "token": TOKEN} if params: req["params"] = params if fields: req["fields"] = fields resp = requests.post(API_URL, json=req) data = resp.json() if data["code"] != 0: raise Exception(f"API error ({data['code']}): {data['msg']}") fields_list = data["data"]["fields"] items = data["data"]["items"] return [dict(zip(fields_list, item)) for item in items] def get_active_contracts(exchange, fut_code): """获取指定交易所、品种仍在交易的合约列表""" contracts = tushare_query( "fut_basic", params={"exchange": exchange, "fut_code": fut_code, "fut_type": "1"}, fields="ts_code,symbol,name,list_date,delist_date", ) today = datetime.now().strftime("%Y%m%d") return [c for c in contracts if not c.get("delist_date") or c["delist_date"] > today] def get_all_contracts(exchange, fut_code): """获取指定交易所、品种的全部合约（含已退市）""" return tushare_query( "fut_basic", params={"exchange": exchange, "fut_code": fut_code, "fut_type": "1"}, fields="ts_code,symbol,name,list_date,delist_date", ) def save_contract_csv(fut_code, contracts): """对每个活跃合约拉取日线行情并写入 data/{fut_code}/{code}.csv""" 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 csv_path = os.path.join(out_dir, csv_name) rows = tushare_query( "fut_daily", params={"ts_code": ts_code}, fields=",".join(FUT_DAILY_FIELDS), ) if not rows: 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']})") time.sleep(0.3) # tushare 频率限制 def init_db(): conn = sqlite3.connect(DB_PATH) conn.execute(""" 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, 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 ) """) conn.commit() conn.close() def sync_contracts_to_db(fut_code, exchange, contracts): conn = sqlite3.connect(DB_PATH) 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")), ) conn.commit() conn.close() print(f" contracts 表写入 {len(contracts)} 条") def sync_csv_to_db(): 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 表") conn.commit() conn.close() if __name__ == "__main__": # 活跃合约 active = get_active_contracts("CZCE", "FG") # 已退市合约：FG1501-FG1512、FG1601-FG1612、FG1701-FG1712、...、FG2601-FG2607 extra_codes = {f"FG{suffix}.ZCE" for suffix in [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 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() print("完成")