SuperAPI AI 工具集

live-feed

数据与分析

查看项目

更新于 10-15 14:06

技能介绍

用途

live-feed 用于在实时 WebSocket 市场数据上搭建指标计算流程。它基于 OpenAlgo WebSocket 流式传输行情数据,并在实时数据上计算技术指标,可选使用 Plotly 进行实时可视化。

参数

将 $ARGUMENTS 解析为:symbol exchange mode

  • $0 = symbol(例如 SBIN、RELIANCE、NIFTY)。默认:SBIN
  • $1 = exchange(例如 NSE、NSE_INDEX)。默认:NSE
  • $2 = mode(例如 ltp、quote、depth、multi)。默认:quote

如果没有参数,则询问用户所需的 symbol 和数据类型。

主要能力

该技能会读取 indicator-expert 规则,尤其是:

  • rules/websocket-feeds.md — WebSocket 连接与订阅
  • rules/data-fetching.md — 用于缓冲区初始化的历史数据

随后按需创建 charts/live/ 目录,创建 {symbol}_live_feed.py,并使用 rules/assets/live_feed/template.py 中的模板。

数据模式

ltp — 最新成交价 + 指标

  • 订阅 LTP 数据流
  • 维护滚动缓冲区(最近 200 个 tick)
  • 在缓冲区上计算 EMA、RSI
  • 打印实时指标值

quote — 完整报价 + 指标

  • 订阅 Quote 数据流
  • 显示 OHLC + LTP + 成交量
  • 在收盘价缓冲区上计算指标
  • 彩色输出(看涨/看跌)

depth — 市场深度分析

  • 订阅 Depth 数据流
  • 显示 L5 买卖盘口
  • 计算买卖价差、订单失衡
  • 显示总买量与总卖量

multi — 多品种数据流

  • 订阅多个品种
  • 显示包含 LTP 和关键指标的观察列表表格
  • 自动刷新显示

使用方式

脚本结构示例:

"""
Real-Time Indicator Feed for {SYMBOL}
Mode: {mode}
"""
import os
import time
import numpy as np
from datetime import datetime, timedelta
from dotenv import find_dotenv, load_dotenv
from openalgo import api, ta

load_dotenv(find_dotenv(), override=False)

SYMBOL = "{symbol}"
EXCHANGE = "{exchange}"

client = api(
    api_key=os.getenv("OPENALGO_API_KEY"),
    host=os.getenv("OPENALGO_HOST", "http://127.0.0.1:5000"),
    verbose=1,
)

# Pre-fetch historical data for buffer initialization
df = client.history(
    symbol=SYMBOL, exchange=EXCHANGE, interval="1m",
    start_date=(datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d"),
    end_date=datetime.now().strftime("%Y-%m-%d"),
)
close_buffer = list(df["close"].values[-200:])

instruments = [{"exchange": EXCHANGE, "symbol": SYMBOL}]

def on_data(data):
    ltp = data["data"].get("ltp")
    if ltp is None:
        return

    close_buffer.append(float(ltp))
    if len(close_buffer) > 200:
        close_buffer.pop(0)

    if len(close_buffer) >= 20:
        arr = np.array(close_buffer, dtype=np.float64)
        ema_val = ta.ema(arr, 20)[-1]
        rsi_val = ta.rsi(arr, 14)[-1] if len(arr) >= 15 else float("nan")

        timestamp = datetime.now().strftime("%H:%M:%S")
        print(f"[{timestamp}] {SYMBOL} LTP:{ltp:>10.2f} | "
              f"EMA(20):{ema_val:>10.2f} | RSI(14):{rsi_val:>6.2f}")

# Connect and subscribe
client.connect()
client.subscribe_ltp(instruments, on_data_received=on_data)

print(f"Streaming {SYMBOL} on {EXCHANGE} — Press Ctrl+C to stop")
try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    print("Stopping feed...")

client.unsubscribe_ltp(instruments)
client.disconnect()

清理

脚本必须:

  • 优雅处理 Ctrl+C
  • 取消订阅所有数据流
  • 断开 WebSocket 连接
  • 打印会话时长和处理过的 bar 数量摘要

详细级别

向用户说明 verbose 选项:

  • verbose=0:静默模式(仅错误)
  • verbose=1:连接与订阅日志
  • verbose=2:所有数据更新(调试模式)

示例用法

/live-feed SBIN NSE ltp
/live-feed NIFTY NSE_INDEX quote
/live-feed SBIN NSE depth
/live-feed multi NSE

适用场景

适用于需要在实时行情数据上计算技术指标的场景,例如实时监控单只股票的 LTP 与 EMA/RSI、查看完整报价与成交量、分析市场深度与订单失衡,或同时跟踪多个品种的观察列表。

限制

资料中未提供关于收费、价格、平台兼容性、性能上限或具体效果的信息。