exsdk 期货 K 线与 Python 指标计算

K 线仅支持期货，通过 watch_kline() 创建固定长度窗口。

创建固定长度窗口#

bars = api.watch_kline(symbol, "1m", data_length=200)

参数	可用值
`period`	`"1m"`、`"5m"`、`"10m"`、`"15m"`、`"30m"`、`"1h"`、`"4h"`、`"1d"`、`"1w"`（周线）、`"mn"`（月线）
`data_length`	1–10000 的整数，默认 200

初始化：同步加载历史窗口，返回后即可调用 data() 读取；初始化可能等待网络。
句柄复用：相同合约、周期、长度复用同一句柄，不同长度返回不同句柄。
期权限制：期权没有 K 线能力，误传期权不会得到有效 K 线。

窗口长度与有效数据#

data() 始终返回 data_length 行。历史数据不足时，前面的行以 timestamp=0、trading_day="" 和数值 NaN 补齐。因此，frame.empty 和 len(frame) 不能用于判断有无行情或有效 K 线数量。

先排除补齐行，再检查计算所需的价格字段：

import math

frame = bars.data()
valid = frame.loc[frame["timestamp"] > 0]
print("有效时间戳行数", len(valid))
if not valid.empty and math.isfinite(valid["close"].iloc[-1]):
    print(valid.iloc[-1].to_dict())

不要将缺失价格填成 0 后参与指标计算。窗口列、类型和量额口径见 K 线数据字典。按日期下载的历史数据不补齐窗口，无数据时返回空表，见历史下载。

时间与末行#

timestamp 为 Unix 毫秒；分钟、小时线标记窗口结束时间，日、周、月线分别标记北京时间交易日、周一、月初零点。

周月线按交易日所属的自然周、自然月划分，trading_day 为周期内最新交易日。

weekly = api.watch_kline(symbol, "1w", data_length=100)
monthly = api.watch_kline(symbol, "mn", data_length=60)

末行数据：最后一行可能仍在形成，后续更新仍可能改变它。

frame.attrs["symbol"]、frame.attrs["period"] 保存合约和周期。实时 K 线按同一时间戳更新，不是每次推送都增加一行；没有新成交时，不保证每个自然时间窗口都有一根 K 线。

计算移动平均#

import math

bars = api.watch_kline(symbol, "1m", data_length=200)
while api.wait_update():
    if bars.is_changing(["timestamp", "close"]):
        frame = bars.data()
        valid = frame.loc[frame["timestamp"] > 0]
        if len(valid) >= 20:
            latest = valid["close"].rolling(20).mean().iloc[-1]
            if math.isfinite(latest):
                print("当前末行 MA20", latest)

这里计算包含当前末行的 MA20，最后 20 行价格存在缺失时结果为 NaN。初始化指标可在循环前读取窗口计算一次；循环中的 is_changing() 只检测后续更新。自行添加的指标列只属于当前快照，下一次读取时需重新计算或由业务层维护。

同时观察行情与 K 线#

下面片段放在已创建 api、选定期货合约 symbol 的 with 内，多个订阅共用一个更新循环：

quote = api.watch_quote(symbol)
bars = api.watch_kline(symbol, "1m", data_length=200)

while api.wait_update():
    if quote.is_changing("last_price"):
        print("最新价", quote.data().last_price)
    if bars.is_changing(["timestamp", "close"]):
        frame = bars.data()
        valid = frame.loc[frame["timestamp"] > 0]
        if not valid.empty:
            print("窗口末行", valid.iloc[-1].to_dict())

先创建所有句柄，再通过一个循环处理各自的变化。需要限制等待时间时，参见有限等待。

按完成 K 线处理一次#

is_changing() 对 K 线只比较末行。新 K 线与前一根收盘价相同时，is_changing("close") 可以是 False；检查 "timestamp" 才能识别末行时间推进。历史行修正而末行不变时，也不能靠它检测整个窗口变化。

下面采用保守规则：看到后一根 K 线后，才处理它之前的 K 线。示例跳过初始化时已存在的历史完成行，从初始化末行开始跟踪；一次推进多根时逐行处理，而非只取倒数第二行。代码替换原有 K 线更新循环：

import math
import time

bars = api.watch_kline(symbol, "1m", data_length=200)
initial = bars.data()
initial = initial.loc[initial["timestamp"] > 0]
last_processed = int(initial.iloc[-2]["timestamp"]) if len(initial) >= 2 else 0

while True:
    frame = bars.data()
    valid = frame.loc[frame["timestamp"] > 0]
    completed = valid.iloc[:-1]
    for bar in completed.itertuples(index=False):
        if bar.timestamp <= last_processed:
            continue
        if not math.isfinite(bar.close):
            raise ValueError("Completed bar has no valid close price")
        print("新完成 K 线", bar.timestamp, bar.close)
        last_processed = int(bar.timestamp)
    if not api.wait_update(deadline=time.time() + 5):
        continue

这是应用层的完成判定，不是服务提供的最终确认标记。休市前的最后一根可能要到下次出现新 K 线才被处理；不能仅凭本地时间认定已收到最终数据。时间戳去重不处理之后对旧行的修正，重启后也需自行保存和恢复进度。

断线恢复只能在当前窗口内补处理；断线过久、遗漏超过窗口长度时，需用历史下载核对缺口。示例只打印数据，实际策略应先完成核对，再恢复交易决策。

K 线与指标