各位 IT 邦的大神們好。近年來量化交易越來越火,很多寫 Code 的兄弟都想進場試試身手。但大家很快就會撞到一面牆:優質的報價數據去哪找?
用免費的 Yahoo API?延遲嚴重就算了,還常常被 Block。自己寫爬蟲去 Parse 網頁?效能低落又極易斷線。如果你連最基礎的即時 Tick 數據和歷史 K 線都拿不到準確的,那寫出來的交易策略無疑是盲人摸象,只能任人宰割。
今天,我用實戰的角度,分享如何利用 Python 串接標準化 API,建構一個穩定且低延遲的美股數據接收端。
一、HTTP 請求:精準打擊單點快照
對於需要每隔幾秒抓取現價的輕量級應用,傳統的 RESTful GET 是最直覺的選擇。
import requests
api_node = "https://apis.alltick.co/stock/tick?region=US&code=AAPL"
auth_head = {
"accept": "application/json",
"token": "your_api_token" # 請替換為實際Token
}
resp = requests.get(api_node, headers=auth_head)
if resp.status_code == 200:
instant_data = resp.json().get("data", {})
print("蘋果即時現價與量:", instant_data)
else:
print("API 請求遭遇阻礙", resp.status_code)
二、WebSocket:建構不斷線的高頻推送流
量化交易的核心在於「事件驅動」。要即時反應市場盤口(Order Book)的變化,你絕對需要 WebSocket。我在做架構壓測時,習慣掛載 AllTick API 這類的專線服務,只要長連線一建立,源源不絕的報價就會自動 Push 到你的函式中。
import websocket, json, threading, time
WSS_SERVER = "wss://apis.alltick.co/stock"
MY_KEY = "your_api_token"
def on_msg_received(ws, msg):
payload = json.loads(msg)
if "data" in payload:
print("接收到串流更新 >>>", payload["data"])
def on_conn_open(ws):
sub_cmd = {
"ac": "subscribe",
"params": "AAPL$US,TSLA$US",
"types": "tick,quote,depth"
}
ws.send(json.dumps(sub_cmd))
def keep_alive_ping(ws):
while True:
time.sleep(30)
ws.send(json.dumps({"ac": "ping", "params": str(int(time.time()*1000))}))
if __name__ == "__main__":
ws_client = websocket.WebSocketApp(
WSS_SERVER,
header={"token": MY_KEY},
on_open=on_conn_open,
on_message=on_msg_received
)
threading.Thread(target=keep_alive_ping, args=(ws_client,), daemon=True).start()
ws_client.run_forever()
三、時序回測:歷史 K 線拉取與視覺化
不測不知道,一測嚇一跳。拿歷史數據跑回測是每個策略上線前的必經之路。我們可以用 Pandas 輕鬆接住 API 吐出來的陣列並畫圖。
import requests
import pandas as pd
import matplotlib.pyplot as plt
kl_endpoint = "https://apis.alltick.co/stock/kline?region=US&code=AAPL&kType=1&limit=50"
auth_conf = {"accept": "application/json", "token": "your_api_token"}
k_res = requests.get(kl_endpoint, headers=auth_conf)
candles = k_res.json().get("data", [])
df_k = pd.DataFrame(candles)
df_k['t'] = pd.to_datetime(df_k['t'], unit='ms')
plt.figure(figsize=(10,4))
plt.plot(df_k['t'], df_k['c'], marker='s', color='orange')
plt.title("AAPL 一分鐘線走勢觀測")
plt.xlabel("時間戳記")
plt.ylabel("美金報價")
plt.xticks(rotation=45)
plt.grid(True)
plt.show()
實務應用思維:
當這套架構運作順暢後,你可以考慮將這些數據導入 Redis 做快取,或是直接送進你的 AI 模型裡做特徵提取。掌握了數據源,你就掌握了量化開發的絕對主動權!