股票监控skill实战：如何构建高可靠性的实时预警系统

3次阅读

共计 2274 个字符，预计需要花费 6 分钟才能阅读完成。

在股票监控场景中，传统轮询方案存在两个致命缺陷：

高延迟：固定间隔的请求可能错过关键价格波动，对于高频交易场景，500ms 的延迟就可能造成数百万损失
资源浪费：无差别轮询所有标的，导致 80% 以上的请求返回无变化数据，某券商实测显示浪费了 73% 的带宽资源

我们的解决方案采用分层架构：

[数据源层] ——WebSocket——> [事件处理器] ——RabbitMQ——> [分析引擎] ——Alarm——> [通知服务]
   ↑                               ↓
[API 降级通道]                [Redis 状态缓存]

关键设计原则：

使用 WebSocket 长连接作为主要数据通道
消息队列实现生产消费解耦
同步 API 作为降级备选方案

import aiohttp
import asyncio

async def fetch_market_data(symbols):
    async with aiohttp.ClientSession() as session:
        tasks = [
            asyncio.create_task(fetch_exchange_data(session, symbol)
            ) for symbol in symbols
        ]
        return await asyncio.gather(*tasks, return_exceptions=True)

# NOTE: 每个交易所 API 需要单独配置连接池参数
async def fetch_exchange_data(session, symbol):
    url = f"https://api.exchange.com/ticker?symbol={symbol}"
    try:
        async with session.get(url, timeout=1.5) as resp:
            resp.raise_for_status()
            return await resp.json()
    except Exception as e:
        logging.error(f"{symbol}数据获取失败: {str(e)}")
        return None

import redis
import time

r = redis.Redis(host='localhost', port=6379)

def check_abnormal_volume(symbol, window=60, threshold=3):
    """
    :param window: 统计窗口(秒)
    :param threshold: 标准差倍数阈值
    """
    now = int(time.time())
    pipe = r.pipeline()

    # 记录当前交易量
    pipe.zadd(f"vol:{symbol}", {now: current_volume})
    # 移除过期数据
    pipe.zremrangebyscore(f"vol:{symbol}", 0, now - window)
    # 获取窗口内数据
    pipe.zrange(f"vol:{symbol}", 0, -1, withscores=True)

    _, _, vol_data = pipe.execute()

    if len(vol_data) < 5:  # 数据不足不触发
        return False

    values = [v for _, v in vol_data]
    mean = sum(values) / len(values)
    std = (sum((x - mean)**2 for x in values) / len(values))**0.5

    return current_volume > mean + threshold * std

采用指数加权移动平均 (EWMA) 调整阈值：

New_Threshold = α * Current_Volatility + (1-α) * Previous_Threshold

其中 α∈(0,1)为平滑系数，建议取值 0.05-0.2

初次断开：立即重连
第二次断开：等待 1 秒 + 随机 0 - 1 秒
后续断开：等待时间按 2^n 指数增长，上限 5 分钟

必须埋点的核心指标：

数据接收延迟(ms)
消息处理队列长度
异常触发频率
API 调用成功率

令牌桶算法控制请求速率
优先保障高价值标的的数据获取
建立交易所黑白名单机制

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_tokens, refill_rate):
        self.tokens = max_tokens
        self.max_tokens = max_tokens
        self.refill_rate = refill_rate  # tokens/sec
        self.last_refill = time.time()

    def consume(self):
        now = time.time()
        elapsed = now - self.last_refill
        self.tokens = min(
            self.max_tokens,
            self.tokens + elapsed * self.refill_rate
        )
        self.last_refill = now

        if self.tokens >= 1:
            self.tokens -= 1
            return True
        return False

可尝试的方向：