fish/finance-talk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

新增行情查询、价差图表和 spread 分析表

Browse Source 
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
vipg
2026-07-18 17:35:05 +08:00
parent 7cf2449b59
commit 010d65504a
4 changed files with 434 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
chart.html
+121
View File
File diff suppressed because one or more lines are too long
chart.py
+149
View File
@@ -0,0 +1,149 @@
"""
生成 spread 数据图表（HTML）。
用法: python3 chart.py
"""
import sqlite3
import json
DB_PATH = "/Users/vipg/Documents/futures-data-warehouse/db/futures.db"
def main():
conn = sqlite3.connect(DB_PATH)
rows = conn.execute(
"SELECT trade_date, spread, main_chg FROM spread ORDER BY trade_date"
).fetchall()
conn.close()
dates = [r[0] for r in rows]
spreads = [r[1] for r in rows]
chgs = [r[2] if r[2] is not None else None for r in rows]
html = f"""<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>玻璃期货 spread 走势</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js@4"></script>
<style>
* {{ margin: 0; padding: 0; box-sizing: border-box; }}
body {{ background: #0f172a; color: #e2e8f0; font-family: system-ui, sans-serif; padding: 24px; }}
h1 {{ font-size: 20px; margin-bottom: 24px; color: #f1f5f9; }}
.chart-box {{ background: #1e293b; border-radius: 12px; padding: 20px; }}
canvas {{ width: 100% !important; height: auto !important; }}
.stats {{ display: flex; gap: 24px; flex-wrap: wrap; margin-top: 24px; }}
.stat {{ background: #1e293b; border-radius: 12px; padding: 16px 20px; min-width: 130px; }}
.stat-label {{ font-size: 12px; color: #94a3b8; }}
.stat-value {{ font-size: 22px; font-weight: 600; margin-top: 4px; }}
.stat-value.pos {{ color: #22c55e; }}
.stat-value.neg {{ color: #ef4444; }}
</style>
</head>
<body>
<h1>玻璃期货 — 合约链差价 &amp; 主力合约日内涨跌</h1>
<div class="chart-box">
<canvas id="chart"></canvas>
</div>
<div class="stats" id="stats"></div>
<script>
const dates = {json.dumps(dates)};
const spreads = {json.dumps(spreads)};
const chgs = {json.dumps(chgs)};
new Chart(document.getElementById('chart'), {{
type: 'bar',
data: {{
labels: dates,
datasets: [
{{
label: '主力日内涨跌',
data: chgs,
yAxisID: 'y1',
backgroundColor: chgs.map(v => v === null ? 'transparent' : v >= 0 ? 'rgba(34,197,94,0.35)' : 'rgba(239,68,68,0.35)'),
borderColor: chgs.map(v => v === null ? 'transparent' : v >= 0 ? '#22c55e' : '#ef4444'),
borderWidth: 0.3,
order: 2,
}},
{{
label: '差价合计',
data: spreads,
yAxisID: 'y',
type: 'line',
borderColor: '#facc15',
backgroundColor: 'rgba(250,204,21,0.06)',
borderWidth: 1.5,
pointRadius: 0,
fill: true,
tension: 0.1,
order: 1,
}},
]
}},
options: {{
responsive: true,
interaction: {{ mode: 'index', intersect: false }},
plugins: {{
legend: {{ labels: {{ color: '#94a3b8', boxWidth: 14, padding: 16 }} }},
tooltip: {{
backgroundColor: '#0f172a',
titleColor: '#f1f5f9',
bodyColor: '#e2e8f0',
borderColor: '#334155',
borderWidth: 1,
callbacks: {{
label: ctx => ctx.parsed.y !== null && ctx.parsed.y !== undefined
? `${{ctx.dataset.label}}: ${{ctx.parsed.y.toFixed(1)}}` : ''
}}
}}
}},
scales: {{
x: {{ ticks: {{ color: '#64748b', maxTicksLimit: 20, font: {{ size: 10 }} }}, grid: {{ color: '#1e293b' }} }},
y: {{
position: 'left',
ticks: {{ color: '#facc15' }},
grid: {{ color: '#334155' }},
title: {{ display: true, text: '差价合计（元/吨）', color: '#facc15' }}
}},
y1: {{
position: 'right',
ticks: {{ color: '#94a3b8' }},
grid: {{ drawOnChartArea: false }},
title: {{ display: true, text: '主力日内涨跌（元/吨）', color: '#94a3b8' }}
}}
}}
}}
}});
// 统计
const pos = spreads.filter(v => v > 0).length;
const neg = spreads.filter(v => v < 0).length;
const avg = spreads.reduce((a,b) => a+b, 0) / spreads.length;
const maxV = Math.max(...spreads);
const minV = Math.min(...spreads);
const totalChg = chgs.filter(v => v !== null).reduce((a,b) => a+b, 0);
const avgChg = totalChg / chgs.filter(v => v !== null).length;
const statsHtml = `
<div class="stat"><div class="stat-label">数据天数</div><div class="stat-value">${{spreads.length}}</div></div>
<div class="stat"><div class="stat-label">平均差价</div><div class="stat-value ${{avg>=0?'pos':'neg'}}">${{avg.toFixed(1)}}</div></div>
<div class="stat"><div class="stat-label">升水天数</div><div class="stat-value pos">${{pos}}</div></div>
<div class="stat"><div class="stat-label">贴水天数</div><div class="stat-value neg">${{neg}}</div></div>
<div class="stat"><div class="stat-label">最大升水</div><div class="stat-value pos">+${{maxV.toFixed(1)}}</div></div>
<div class="stat"><div class="stat-label">最大贴水</div><div class="stat-value neg">${{minV.toFixed(1)}}</div></div>
<div class="stat"><div class="stat-label">主力日均涨跌</div><div class="stat-value ${{avgChg>=0?'pos':'neg'}}">${{avgChg.toFixed(2)}}</div></div>
`;
document.getElementById('stats').innerHTML = statsHtml;
</script>
</body>
</html>"""
out_path = "/Users/vipg/Documents/futures-data-warehouse/chart.html"
with open(out_path, "w") as f:
f.write(html)
print(f"已生成: {out_path}")
if __name__ == "__main__":
main()
quote.py
+87
View File
@@ -0,0 +1,87 @@
"""
查询指定交易日玻璃期货主力合约行情。
用法: python3 quote.py 20260717
"""
import sqlite3
import sys
from datetime import datetime
DB_PATH = "/Users/vipg/Documents/futures-data-warehouse/db/futures.db"
def main():
if len(sys.argv) < 2:
print("用法: python3 quote.py YYYYMMDD")
sys.exit(1)
date_str = sys.argv[1]
try:
datetime.strptime(date_str, "%Y%m%d")
except ValueError:
print(f"日期格式错误: {date_str}，应为 YYYYMMDD")
sys.exit(1)
conn = sqlite3.connect(DB_PATH)
count = conn.execute(
"SELECT COUNT(*) FROM daily WHERE trade_date=?", (date_str,)
).fetchone()[0]
if count == 0:
import urllib.request, json
req_data = json.dumps({
"api_name": "trade_cal",
"token": "76efd8465f9f2591aa42a385268e06acf6b80b7a15be2267ad2281b7",
"params": {"exchange": "CZCE", "start_date": date_str, "end_date": date_str},
"fields": "cal_date,is_open",
}).encode()
resp = json.loads(urllib.request.urlopen(
urllib.request.Request("https://api.tushare.pro", data=req_data,
headers={"Content-Type": "application/json"})
).read())
is_open = resp.get("data", {}).get("items", [[None, None]])[0][1]
if is_open == 1:
print(f"{date_str} 是交易日，但数据库尚无数据（需先运行 update.py 拉取）")
elif is_open == 0:
print(f"{date_str} 非交易日")
else:
print(f"{date_str} 无数据")
conn.close()
sys.exit(1)
rows = conn.execute("""
SELECT d.ts_code, d.open, d.high, d.low, d.close
FROM daily d
JOIN contracts c ON d.ts_code = c.ts_code
WHERE d.trade_date = ?
AND c.delist_date > ?
AND d.oi > 0
ORDER BY c.delist_date
""", (date_str, date_str)).fetchall()
if not rows:
print(f"{date_str} 无活跃合约数据")
conn.close()
sys.exit(1)
print(f"\n{date_str} 玻璃期货")
print(f"{'合约':>12} 收盘价 差价")
print("-" * 34)
prev = None
total_diff = 0.0
for r in rows:
close = float(r[4]) if r[4] is not None and r[4] != "" else None
close_str = f"{close:.1f}" if close else "-"
diff = ""
if close is not None and prev is not None:
d = close - prev
diff = f"{d:+.1f}"
total_diff += d
print(f"{r[0]:>12} {close_str:>7} {diff:>6}")
if close is not None:
prev = close
print("-" * 34)
print(f"{'差价合计':>20} {total_diff:+.1f}")
conn.close()
if __name__ == "__main__":
main()
update.py
+77 -49
View File
@@ -20,40 +20,22 @@ 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): def tushare_query(api_name, params=None, fields=None):
req = {"api_name": api_name, "token": TOKEN} req = {"api_name": api_name, "token": TOKEN}
if params: if params:
req["params"] = params req["params"] = params
if fields: if fields:
req["fields"] = fields req["fields"] = fields
resp = requests.post(API_URL, json=req) resp = requests.post(API_URL, json=req)
data = resp.json() data = resp.json()
if data["code"] != 0: if data["code"] != 0:
raise Exception(f"API error ({data['code']}): {data['msg']}") raise Exception(f"API error ({data['code']}): {data['msg']}")
fields_list = data["data"]["fields"] fields_list = data["data"]["fields"]
items = data["data"]["items"] items = data["data"]["items"]
return [dict(zip(fields_list, item)) for item in items] return [dict(zip(fields_list, item)) for item in items]
def is_trading_day(exchange="CZCE"): def is_trading_day(exchange="CZCE"):
"""检查指定交易所今天是否开市"""
today = datetime.now().strftime("%Y%m%d") today = datetime.now().strftime("%Y%m%d")
rows = tushare_query( rows = tushare_query(
"trade_cal", "trade_cal",
@@ -62,7 +44,7 @@ def is_trading_day(exchange="CZCE"):
) )
if rows: if rows:
return rows[0]["is_open"] == "1" return rows[0]["is_open"] == "1"
return True # 查不到时保守放行 return True
# ── 合约列表 ────────────────────────────────── # ── 合约列表 ──────────────────────────────────
@@ -88,18 +70,12 @@ def get_all_contracts(exchange, fut_code):
# ── 增量更新 CSV ────────────────────────────── # ── 增量更新 CSV ──────────────────────────────
def update_contract_csv(fut_code, contract): def update_contract_csv(fut_code, contract):
"""
增量更新单份合约 CSV。
从已有数据的最后交易日重新拉取（含该日），实现补漏 + 修正。
返回新增/修改的行，或 None（无变化）。
"""
ts_code = contract["ts_code"] ts_code = contract["ts_code"]
csv_name = ts_code.split(".")[0] + ".csv" csv_name = ts_code.split(".")[0] + ".csv"
out_dir = os.path.join(DATA_DIR, fut_code) out_dir = os.path.join(DATA_DIR, fut_code)
csv_path = os.path.join(out_dir, csv_name) csv_path = os.path.join(out_dir, csv_name)
os.makedirs(out_dir, exist_ok=True) os.makedirs(out_dir, exist_ok=True)
# 读取已有数据，获取最后交易日
existing = [] existing = []
if os.path.exists(csv_path): if os.path.exists(csv_path):
with open(csv_path) as f: with open(csv_path) as f:
@@ -108,7 +84,6 @@ def update_contract_csv(fut_code, contract):
last_date = existing[-1]["trade_date"] if existing else None last_date = existing[-1]["trade_date"] if existing else None
# 从最后交易日开始拉取（包含该日，以便修正）
params = {"ts_code": ts_code} params = {"ts_code": ts_code}
if last_date: if last_date:
params["start_date"] = last_date params["start_date"] = last_date
@@ -120,7 +95,6 @@ def update_contract_csv(fut_code, contract):
rows.sort(key=lambda r: r["trade_date"]) rows.sort(key=lambda r: r["trade_date"])
if not existing: if not existing:
# 全新合约
with open(csv_path, "w", newline="") as f: with open(csv_path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS)
w.writeheader() w.writeheader()
@@ -128,14 +102,10 @@ def update_contract_csv(fut_code, contract):
return rows return rows
if rows[0]["trade_date"] == last_date: if rows[0]["trade_date"] == last_date:
# 最后交易日可能有修正 → 替换最后一行再追加
existing = existing[:-1] existing = existing[:-1]
merged = existing + rows merged = existing + rows
elif rows[0]["trade_date"] > last_date:
# 纯新增数据
merged = existing + rows
else: else:
merged = existing + rows # 安全兜底 merged = existing + rows
with open(csv_path, "w", newline="") as f: with open(csv_path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS)
@@ -146,7 +116,6 @@ def update_contract_csv(fut_code, contract):
def update_all_contracts(fut_code, contracts): def update_all_contracts(fut_code, contracts):
"""批量增量更新，返回所有有变化的行"""
all_new = [] all_new = []
for c in contracts: for c in contracts:
code = c["ts_code"].split(".")[0] code = c["ts_code"].split(".")[0]
@@ -165,27 +134,21 @@ def update_all_contracts(fut_code, contracts):
def fetch_all_csv(fut_code, contracts): def fetch_all_csv(fut_code, contracts):
out_dir = os.path.join(DATA_DIR, fut_code) out_dir = os.path.join(DATA_DIR, fut_code)
os.makedirs(out_dir, exist_ok=True) os.makedirs(out_dir, exist_ok=True)
for c in contracts: for c in contracts:
ts_code = c["ts_code"] ts_code = c["ts_code"]
csv_name = ts_code.split(".")[0] + ".csv" csv_name = ts_code.split(".")[0] + ".csv"
csv_path = os.path.join(out_dir, csv_name) csv_path = os.path.join(out_dir, csv_name)
rows = tushare_query( rows = tushare_query(
"fut_daily", "fut_daily", params={"ts_code": ts_code}, fields=",".join(FUT_DAILY_FIELDS),
params={"ts_code": ts_code},
fields=",".join(FUT_DAILY_FIELDS),
) )
if not rows: if not rows:
print(f" {csv_name:12s} 无数据，跳过") print(f" {csv_name:12s} 无数据，跳过")
continue continue
rows.sort(key=lambda r: r["trade_date"]) rows.sort(key=lambda r: r["trade_date"])
with open(csv_path, "w", newline="") as f: with open(csv_path, "w", newline="") as f:
w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS) w = csv.DictWriter(f, fieldnames=FUT_DAILY_FIELDS)
w.writeheader() w.writeheader()
w.writerows(rows) 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)
@@ -196,9 +159,8 @@ def init_db():
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH)
conn.execute(""" conn.execute("""
CREATE TABLE IF NOT EXISTS daily ( CREATE TABLE IF NOT EXISTS daily (
ts_code TEXT NOT NULL, ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
trade_date TEXT NOT NULL, pre_close REAL, pre_settle REAL,
pre_close REAL, pre_settle REAL,
open REAL, high REAL, low REAL, close REAL, settle REAL, open REAL, high REAL, low REAL, close REAL, settle REAL,
change1 REAL, change2 REAL, change1 REAL, change2 REAL,
vol REAL, amount REAL, oi REAL, oi_chg REAL, vol REAL, amount REAL, oi REAL, oi_chg REAL,
@@ -213,6 +175,18 @@ def init_db():
fut_code TEXT, list_date TEXT, delist_date TEXT fut_code TEXT, list_date TEXT, delist_date TEXT
) )
""") """)
conn.execute("""
CREATE TABLE IF NOT EXISTS spread (
trade_date TEXT PRIMARY KEY,
spread REAL,
main_ts_code TEXT,
main_chg REAL
)
""")
try:
conn.execute("ALTER TABLE spread ADD COLUMN main_chg REAL")
except sqlite3.OperationalError:
pass
conn.commit() conn.commit()
conn.close() conn.close()
@@ -232,7 +206,6 @@ def sync_contracts_to_db(fut_code, exchange, contracts):
def sync_rows_to_db(rows): def sync_rows_to_db(rows):
"""只同步有变化的行到 daily 表（增量更新用）"""
if not rows: if not rows:
return return
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH)
@@ -247,7 +220,6 @@ def sync_rows_to_db(rows):
def sync_all_csv_to_db(): def sync_all_csv_to_db():
"""全量同步所有 CSV 到 daily 表（初始化用）"""
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH)
cursor = conn.cursor() cursor = conn.cursor()
for root, _, files in os.walk(DATA_DIR): for root, _, files in os.walk(DATA_DIR):
@@ -270,6 +242,63 @@ def sync_all_csv_to_db():
conn.close() conn.close()
def sync_spread():
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
dates = conn.execute("SELECT DISTINCT trade_date FROM daily ORDER BY trade_date").fetchall()
count = 0
for d in dates:
date_str = d["trade_date"]
rows = conn.execute("""
SELECT d.ts_code, d.close, d.open, d.vol
FROM daily d
JOIN contracts c ON d.ts_code = c.ts_code
WHERE d.trade_date = ?
AND c.delist_date > ?
AND d.oi > 0
ORDER BY c.delist_date
""", (date_str, date_str)).fetchall()
if len(rows) < 2:
continue
closes = []
main_ts = None
main_vol = -1
main_chg = None
for r in rows:
try:
closes.append(float(r["close"]))
except (TypeError, ValueError):
closes.append(None)
if r["ts_code"][4:6] in ("01", "05", "09"):
v = r["vol"]
if v is not None and v > main_vol:
main_vol = v
main_ts = r["ts_code"]
try:
c = float(r["close"])
o = float(r["open"])
main_chg = c - o
except (TypeError, ValueError):
main_chg = None
closes = [c for c in closes if c is not None]
if len(closes) < 2:
continue
spread = closes[-1] - closes[0]
conn.execute(
"INSERT OR REPLACE INTO spread VALUES (?, ?, ?, ?)",
(date_str, spread, main_ts, main_chg),
)
count += 1
conn.commit()
conn.close()
print(f" spread 表更新 {count} 个交易日")
# ── 主入口 ──────────────────────────────────── # ── 主入口 ────────────────────────────────────
FG_EXTRA_CODES = {f"FG{suffix}.ZCE" for suffix in FG_EXTRA_CODES = {f"FG{suffix}.ZCE" for suffix in
@@ -290,7 +319,6 @@ FG_EXTRA_CODES = {f"FG{suffix}.ZCE" for suffix in
if __name__ == "__main__": if __name__ == "__main__":
full_init = "--init" in sys.argv full_init = "--init" in sys.argv
init_db() init_db()
active = get_active_contracts("CZCE", "FG") active = get_active_contracts("CZCE", "FG")
@@ -301,17 +329,16 @@ if __name__ == "__main__":
contracts = active + extra contracts = active + extra
print(f"全量初始化: {len(contracts)} 个合约") print(f"全量初始化: {len(contracts)} 个合约")
for c in contracts: for c in contracts:
code = c["ts_code"].split(".")[0] print(f" {c['ts_code'].split('.')[0]:8s} 上市:{c['list_date']} 退市:{c.get('delist_date','-')}")
print(f" {code:8s} 上市:{c['list_date']} 退市:{c.get('delist_date','-')}")
fetch_all_csv("FG", contracts) fetch_all_csv("FG", contracts)
sync_contracts_to_db("FG", "CZCE", contracts) sync_contracts_to_db("FG", "CZCE", contracts)
sync_all_csv_to_db() sync_all_csv_to_db()
sync_spread()
else: else:
if not is_trading_day(): if not is_trading_day():
print("今天非交易日，无需更新") print("今天非交易日，无需更新")
print("完成") print("完成")
sys.exit(0) sys.exit(0)
print(f"增量更新: {len(active)} 个活跃合约") print(f"增量更新: {len(active)} 个活跃合约")
new_rows = update_all_contracts("FG", active) new_rows = update_all_contracts("FG", active)
if new_rows: if new_rows:
@@ -320,4 +347,5 @@ if __name__ == "__main__":
else: else:
print(" 无新数据") print(" 无新数据")
sync_spread()
print("完成") print("完成")