python
每日 OHLCV 收集器。默认执行需要明确的本地 fixture。
import argparse
import json
import logging
import os
import re
import sqlite3
import time
from datetime import date, datetime, timezone
from decimal import Decimal
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
def fetch(symbol):
key = os.environ["ALPHAVANTAGE_API_KEY"]
query = urlencode(dict(function="TIME_SERIES_DAILY", symbol=symbol,
outputsize="compact", datatype="json", apikey=key))
for attempt in range(3):
try:
request = Request("https://www.alphavantage.co/query?" + query,
headers={"User-Agent": "DailyDataCollector/1.0"})
with urlopen(request, timeout=30) as response:
return json.load(response)
except HTTPError as exc:
# Never print exc or its URL: the query contains the API key.
if exc.code not in (429, 502, 503, 504) or attempt == 2:
raise RuntimeError(f"源HTTP故障: {exc.code}") from None
delay = exc.headers.get("Retry-After", "")
if delay:
# Schedule a later run; never retry earlier than requested.
raise RuntimeError("速率限制; 使用Retry-After重新计划") from None
except (URLError, TimeoutError):
if attempt == 2:
raise RuntimeError("源在三次尝试后不可用") from None
time.sleep(2 ** attempt)
def normalize(payload, symbol, currency, exchange):
if not re.fullmatch(r"[A-Z]{3}", currency) or not exchange.strip():
raise ValueError("提供有效的货币代码和交易所标识符")
if payload.get("Meta Data", {}).get("2. Symbol", "").upper() != symbol.upper():
raise ValueError("缺失或不匹配的提供者符号")
series = payload.get("Time Series (Daily)")
if not isinstance(series, dict) or not series:
raise ValueError("没有每日蜡烛; 检查账户限额或源错误")
observed = datetime.now(timezone.utc).isoformat()
rows = []
for session, bar in sorted(series.items()):
date.fromisoformat(session)
prices = [Decimal(bar[f"{i}. {name}"]) for i, name in
((1, "开盘"), (2, "最高"), (3, "最低"), (4, "收盘"))]
if any(not p.is_finite() or p <= 0 for p in prices):
raise ValueError(f"无效的OHLC: {session}")
opening, high, low, close = prices
if low > min(opening, close) or high < max(opening, close) or low > high:
raise ValueError(f"不一致的蜡烛: {session}")
volume = int(bar["5. volume"])
if volume < 0:
raise ValueError("负值交易量")
rows.append((symbol, exchange, currency, "raw_daily", session,
*(str(p) for p in prices), volume, "alpha_vantage", observed))
return rows
def save(rows, database):
with sqlite3.connect(database) as db:
db.execute("""CREATE TABLE IF NOT EXISTS bars(
symbol TEXT, exchange TEXT, currency TEXT, basis TEXT, session_date TEXT,
open TEXT, high TEXT, low TEXT, close TEXT, volume INTEGER,
source TEXT, observed_at TEXT,
PRIMARY KEY(symbol,exchange,currency,basis,session_date))""")
db.executemany("""INSERT INTO bars VALUES(?,?,?,?,?,?,?,?,?,?,?,?)
ON CONFLICT(symbol,exchange,currency,basis,session_date) DO UPDATE SET
open=excluded.open,high=excluded.high,low=excluded.low,
close=excluded.close,volume=excluded.volume,observed_at=excluded.observed_at
""", rows)
def main():
parser = argparse.ArgumentParser()
parser.add_argument("--symbol", default="IBM")
parser.add_argument("--currency", required=True)
parser.add_argument("--exchange", required=True)
parser.add_argument("--fixture")
parser.add_argument("--live", action="store_true")
parser.add_argument("--db", default="stocks.sqlite")
args = parser.parse_args()
if bool(args.fixture) == args.live:
parser.error("请选择--fixture或--live中的一个")
payload = json.loads(Path(args.fixture).read_text()) if args.fixture else fetch(args.symbol)
rows = normalize(payload, args.symbol, args.currency, args.exchange)
save(rows, args.db)
logging.info("accepted_bars=%s latest_session=%s", len(rows), rows[-1][4])
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
try:
main()
except (ValueError, KeyError, RuntimeError, OSError) as exc:
logging.error("collector_failed: %s", type(exc).__name__)
raise SystemExit(1)
