4、合约信息查询：查询所有合约、查询指定合约、合约状态与交易时间、合约信息缓存策略

合约信息查询，说白了就是交易系统的「地图」。

你连地图都拿不准，怎么敢下单？

我个人习惯，在系统启动的第一时间，先把合约信息拉下来。不是查一次就完事，而是建立一套缓存机制。为什么？因为交易所的合约状态是动态的——临时停牌、调整保证金、修改最小变动价位……这些变化你如果没感知到，订单可能直接废掉。

4.1 查询所有合约

大部分交易所都提供「查询全部合约」的接口。比如 CTP 的 QryInstrument，或者 Binance 的 GET /api/v3/exchangeInfo。

我一般这么处理：

def fetch_all_contracts(exchange):
    """
    拉取全量合约信息
    返回 dict: {symbol: contract_info}
    """
    if exchange == 'ctp':
        req = QryInstrumentField()
        # 发送请求，等待回调
        return self._qry_instrument(req)
    elif exchange == 'binance':
        resp = requests.get('https://api.binance.com/api/v3/exchangeInfo')
        data = resp.json()
        return {s['symbol']: s for s in data['symbols']}
    else:
        raise ValueError(f'不支持的交易所: {exchange}')

这里有个坑——全量查询通常有频率限制。我曾经在实盘环境里，启动时一口气拉了 5 次全量，直接被交易所封了 IP 10 分钟。嗯，后来我学乖了，启动时只拉一次，然后靠增量更新。

4.2 查询指定合约

有时候你不需要全部合约，只想查某一个。比如用户输入了「BTCUSDT」，你只需要验证这个合约是否存在、当前能否交易。

def get_contract(symbol: str, cache: dict) -> dict:
    """
    从缓存中获取指定合约信息
    如果缓存中没有，尝试从交易所拉取
    """
    if symbol in cache:
        return cache[symbol]
    
    # 缓存未命中，拉取单个合约
    # 注意：部分交易所不支持单合约查询，只能全量拉取
    info = fetch_single_contract(symbol)
    if info:
        cache[symbol] = info
    return info

我个人习惯，在查询指定合约时，先查缓存。缓存没有，再走网络请求。为什么？因为用户可能反复查询同一个合约，每次都走网络，延迟高不说，还容易被限流。

4.3 合约状态与交易时间

合约状态，是订单管理的「红绿灯」。

你想想看，如果一个合约处于「停牌」状态，你发个限价单进去，大概率被拒。或者处于「集合竞价」阶段，你发个市价单，可能直接给你个废单。

常见的合约状态包括：

交易时间这块，更得小心。不同交易所、不同品种，交易时间千差万别。A 股是 9:30-15:00，期货有夜盘，币圈是 7×24 小时但每个品种的结算时间不同。

我曾经踩过一个坑：某个期货合约在夜盘 21:00 开盘，我写死了 9:00 开盘。结果夜盘时系统一直报「非交易时间」，订单发不出去。后来我改成从交易所接口动态获取交易时间，才彻底解决。

4.4 合约信息缓存策略

缓存策略，是合约查询模块的核心。我见过不少系统，每次查询都走网络，结果延迟高、容易被限流。也见过缓存永不过期的，合约状态变了都不知道。

我个人推荐「两级缓存 + 定时刷新」的策略：

1. 本地内存缓存：用 dict 或 LRU 缓存，TTL 设为 30 秒。查询时先查内存，命中直接返回。
2. 本地文件/数据库缓存：内存未命中时，查本地持久化存储。TTL 设为 1 小时。
3. 网络请求：两级缓存都未命中，才走网络拉取。拉取后更新两级缓存。

代码实现大概是这样的：

class ContractCache:
    def __init__(self):
        self.memory_cache = {}  # {symbol: (info, expire_time)}
        self.file_cache = {}    # 从本地文件加载
        self.ttl_memory = 30    # 秒
        self.ttl_file = 3600    # 秒
    
    def get(self, symbol):
        # 1. 查内存缓存
        if symbol in self.memory_cache:
            info, expire = self.memory_cache[symbol]
            if time.time() < expire:
                return info
            else:
                del self.memory_cache[symbol]
        
        # 2. 查文件缓存
        info = self._load_from_file(symbol)
        if info:
            self.memory_cache[symbol] = (info, time.time() + self.ttl_memory)
            return info
        
        # 3. 网络请求
        info = self._fetch_from_exchange(symbol)
        if info:
            self.memory_cache[symbol] = (info, time.time() + self.ttl_memory)
            self._save_to_file(symbol, info)
        return info

这里有个细节：定时刷新。我一般会启动一个后台线程，每 30 秒拉一次全量合约信息，更新缓存。这样即使某个合约状态变了，最多 30 秒就能感知到。

最后，画一张图总结一下合约信息查询的完整流程：

这张图把整个流程串起来了。你想想看，如果没有缓存策略，每次查询都走网络，系统压力得有多大？有了两级缓存，90% 以上的查询在内存中就解决了，延迟从几十毫秒降到微秒级。

嗯，合约信息查询这块，说白了就是「快」和「准」的平衡。缓存让你快，定时刷新让你准。两者结合，才能在实盘环境中游刃有余。