解决方案 · 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、企业微信、钉钉。
可以存储数据吗？
为内部分析存储数据很常见；若存储的数据要展示给第三方，请确认套餐的转发条款。