解決方案 · AI 與數據
AI、數據倉庫與監控告警的行情方案
現代投研體系依賴存儲的時間序列和實時信號。iTick 可回填多年K線、把每個 Tick 寫入時序數據庫、為大模型提供真實價格而非過時的訓練數據，並在行情異動時第一時間通知團隊。
- 定時同步K線 + 實時 Tick 落庫
- 支持 ClickHouse、InfluxDB、TimescaleDB、Parquet
- 實時價格與指標作為大模型和智能體的工具
- 突破、波動、大單告警推送到聊天軟件
# WebSocket -> ClickHouse real-time ingestion (batch inserts)
import json, time, websocket, clickhouse_connect # pip install clickhouse-connect
db = clickhouse_connect.get_client(host="localhost")
db.command("""CREATE TABLE IF NOT EXISTS ticks (
sym String, ts DateTime64(3), price Float64, vol Float64
) ENGINE = MergeTree ORDER BY (sym, ts)""")
buf = []
def on_message(ws, msg):
d = json.loads(msg).get("data") or {}
if "ld" in d:
buf.append((d["s"], d.get("t", int(time.time() * 1000)) / 1000, d["ld"], d.get("v", 0)))
if len(buf) >= 500:
db.insert("ticks", buf, column_names=["sym", "ts", "price", "vol"]); buf.clear()
websocket.WebSocketApp("wss://api.itick.org/stock", header={"token": "YOUR_API_TOKEN"},
on_open=lambda ws: ws.send(json.dumps({"ac": "subscribe", "params": "AAPL$US,700$HK", "types": "tick"})),
on_message=on_message).run_forever(ping_interval=20)Tick可存儲顆粒度
多年K線歷史
JSON易於模型與管道處理
6大資產類別
三類數據產品
採集、智能與告警，基於同一數據源。
行情數據湖 / 數據倉庫
定時同步歷史K線，WebSocket 實時寫入 ClickHouse、InfluxDB 或湖倉。
AI 行情大模型與智能投研
為大模型提供時序特徵與指標，用於行情問答、標的篩選與異動解讀。
價格告警與機器人
價格突破、波動率異常、盤口大單推送到企業微信、Telegram、釘釘。
特徵工程
基於一致的 OHLCV 與 Tick 計算收益率、波動率、成交分佈與指標。
適配智能體
簡單的 REST 調用，適合函數調用與 MCP 類集成。
跨市場信號
在一個模型或看板中融合美股、港股、A 股、外匯與加密貨幣。
數據管道
從原始行情到洞察與行動。
- 1
歷史回填
定時通過 REST 同步K線歷史。
- 2
實時入庫
WebSocket Tick 批量寫入時序庫。
- 3
建模分析
特徵、指標與大模型工具讀取存儲與實時報價。
- 4
觸發行動
告警推送到聊天軟件、看板或交易系統。
入庫、推理、告警
三個模塊各一段簡短示例。
- 免費獲取 API Token
- 創建 Tick 表並開始入庫
- 在其上疊加大模型工具與告警規則
# WebSocket -> ClickHouse real-time ingestion (batch inserts)
import json, time, websocket, clickhouse_connect # pip install clickhouse-connect
db = clickhouse_connect.get_client(host="localhost")
db.command("""CREATE TABLE IF NOT EXISTS ticks (
sym String, ts DateTime64(3), price Float64, vol Float64
) ENGINE = MergeTree ORDER BY (sym, ts)""")
buf = []
def on_message(ws, msg):
d = json.loads(msg).get("data") or {}
if "ld" in d:
buf.append((d["s"], d.get("t", int(time.time() * 1000)) / 1000, d["ld"], d.get("v", 0)))
if len(buf) >= 500:
db.insert("ticks", buf, column_names=["sym", "ts", "price", "vol"]); buf.clear()
websocket.WebSocketApp("wss://api.itick.org/stock", header={"token": "YOUR_API_TOKEN"},
on_open=lambda ws: ws.send(json.dumps({"ac": "subscribe", "params": "AAPL$US,700$HK", "types": "tick"})),
on_message=on_message).run_forever(ping_interval=20)AI 與數據常見問題
行情數據適合用什麼數據庫？
ClickHouse、InfluxDB、TimescaleDB 等列式或時序庫能很好地處理 Tick 量級，按幾百行批量寫入效率更高。
iTick 能和大模型、AI 智能體一起用嗎？
可以。把報價與K線接口暴露為工具，模型即可獲取實時價格而非依賴訓練數據；iTick 也提供自己的 AI 金融智能體。
告警可以推送到哪些軟件？
任何支持 Webhook 或機器人 API 的軟件，包括 Telegram、企業微信、釘釘。
可以存儲數據嗎？
為內部分析存儲數據很常見；若存儲的數據要展示給第三方，請確認套餐的轉發條款。