吉宽量化 帮助文档（本地版）

本文档服务于最终用户 + 软件维护者。

所有 API 以 统一接口规范 为权威来源：上层策略 / 选股 / 回测代码只能调用文档中列出的函数，不得绕过接口直接访问数据源或存储文件。

快速导航

按帮助窗口目录顺序排列；路径相对于本文件夹。

文档约定

• 示例代码用 Python（吉宽量化脚本语言）。
• 参数名保持英文，说明用中文。
• 返回值列中，"列" 指 DataFrame 的列。
• 格式标准.md 是软件内部的权威契约 —— 任何数据源适配器、任何新接的数据源，输出格式必须完全对齐。

AI 助手 — 让 AI 帮你写策略代码

概述

AI 助手位于代码编辑器右侧（工具栏 🔌 按钮开关），背后是 AITeam 网站的大模型。它不只是问答：AI 可以直接读取公式库里的策略文件、修改代码、做语法检查，改动经你确认后才会写入文件。

开始使用

1. 登录：点击 AI 助手顶部的用户头像（白色 = 未登录，彩色 = 已登录），浏览器中完成 AITeam 授权。
2. 选角色：左下拉选择 AI 角色（AITeam 雇员）。
3. 选模型：右下拉选择大模型（列表来自你的 AITeam 账号）。
4. 输入需求，回车或点 ➤ 发送。

怎么让 AI 干活（示例提问）

AI 需要读文件、改文件时会自动执行，消息流里会出现工具卡片：

• 🔧 read_file 公式库/策略类/Demo.py — ✓ 86 行（读取成功）
• 🔧 patch_file 公式库/策略类/Demo.py — ✓ 已写入（用户已确认）
• 🔧 delete_file userdata/旧数据.csv — ✓ 已删除（用户已确认）
• 🔧 write_file ... — ✗ 用户拒绝了本次修改

历史对话

点击 AI 助手右上角的 🕘 图标打开历史对话窗口：

• 列表显示你账号下的所有对话组（组名 + 时间，置顶优先、新组在前），与 AITeam 网站端完全同步
• 双击任意一组即可恢复：该组消息重新排版显示（AI 回复保持 Markdown 格式），对话上下文同步重建
• 恢复后继续输入，新消息自动接回原对话组；同一组里换角色、换模型都可以，服务端自动归档
• 组名由 AITeam 平台自动生成；网站端的对话历史页可随时回看完整记录（包括 AI 调用工具的过程）

修改与删除确认（重要）

AI 每次写入或删除文件之前都会弹出确认窗口，你可以逐项核对：

• 修改：弹出"AI 修改确认"窗口，红色行 = 将被删除的原文，绿色行 = 替换后的新内容
• 删除：弹出删除卡片，显示文件路径、行数/字节数和"删除后不可恢复"警告
• 顶部 [✓ 全部接受] [✕ 全部拒绝] 一次处理所有文件；每个文件也可以单独接受或拒绝
• 只有点了接受的内容才会真正执行；点窗口右上角 × 等同于全部拒绝
• 拒绝后 AI 会收到"用户拒绝"，通常会换一种方案再提交
• 一个任务里 AI 最多自动执行 8 轮，超过会停下等你接管

安全机制

1. 范围受限：

• 读/写：AI 只能读写公式库目录内的 .py 文件，无法越出公式库
• 删除：仅允许删除公式库内的 .py 文件和用户数据区 userdata\ 内的文件；软件自身的程序文件一律删不到

1. 语法检查：写入前强制通过语法检查，语法不过的代码根本不会来打扰你确认。
2. 确认执行：所有写入和删除必须经你确认，未确认不落盘。
3. 自动备份：接受修改后，原文件自动备份到 config/ai_backup/（保留原目录结构）；如果改错了，可从对应备份文件复制恢复。删除不备份——删除卡片会明确警告"不可恢复"，请确认前看清文件名。

模型与费用

• 模型列表来自 AITeam 账号，部分模型免费（如 glm-5.3 / glm-5.3-flash）。
• AI 干活（读写文件、修改代码）是多轮调用模型的过程，每一轮都会消耗吉豆。需求说得越明确（文件名、要改成什么），轮次越少、越省。
• 出现"当前模型不支持工具调用"提示时，请在模型下拉换用 glm-5.2、deepseek-v4 等已验证模型。
• 提示"吉豆不足"或"会员未授权/过期"时，请到 AITeam 网站兑换吉豆或续费会员后再使用。
• 每一轮对话都会存档到你 AITeam 账号的网站历史页，可随时回看。

常见问题

Q：AI 说我拒绝了它的修改，但我没看到弹窗？

确认窗口可能被其他窗口挡住，请检查任务栏；关闭该窗口等同全部拒绝。

Q：AI 改错了怎么恢复？

打开 config/ai_backup/ 下对应路径的备份文件，复制回原位置即可。每次接受修改都会覆盖保存"修改前"的原文。

Q：为什么普通聊天也变慢/计费变多？

带工具能力后，AI 可能先读几个文件再回答（每次调用计一轮吉豆）。纯问答类问题影响很小；若想完全省掉，问完即止、避免让它大范围翻文件。

Q：AI 想新建文件会覆盖我已有的文件吗？

不会静默覆盖——任何写入（含新建、整文件覆写）都要经确认窗；覆盖已有文件时 diff 会完整展示改动前后对比。

沙箱与用户代码规范

策略 / 选股 / 指标（AI 目录下的用户代码）运行在吉宽量化的防篡改沙箱中。

沙箱的目标只有一个：保护软件的架构和运算逻辑不被用户代码改动，保证所有策略在同一套公平、稳定的数据环境里运行。

你可以做什么（绝大多数场景不受影响）

你不可以做什么（沙箱违规，会被拦截）

报错样式与处理

加载前（AST 预检），报出文件与行号，策略/指标拒绝加载：

沙箱拦截：d:\stock\AI\策略类\xxx.py 第 12 行：使用了 sys.modules（禁止替换软件模块），已拒绝加载

运行中，抛出 SandboxViolation（PermissionError 子类），上层统一捕获打印：

沙箱拦截：用户代码不允许修改软件模块（模块 symbol_io 的属性 getSymbolData）——数据读取请使用 api 接口

处理方式：按报错内容删掉违规代码。数据获取请改用 api.* 统一接口；需要保存结果请写到用户数据区 userdata\（推荐，见 文件函数/文件函数.md）或软件目录之外。

已知局限（如实告知）

• 纯软件层面防不住蓄意的底层逃逸（C 扩展、gc 遍历等）；沙箱目标是防误用与普通篡改，不是对抗蓄意攻击者
• 用户代码先取得软件类对象引用再改类属性（f = symbol_io.Foo; f.bar = 1），静态预检与守卫机制覆盖不到
• .pyd 编译文件无法做 AST 预检（运行时守卫仍然生效）
• 用户代码里 exec/eval 动态生成的代码，静态模式查不到，但运行时仍受守卫机制约束

图表绘图 API 说明

依据源码整理：candleplot.py（CandlePlot 类）。所有函数名、参数、默认值均与当前运行代码一致。

绘图 API 分两类：

- 第一类：画指标 —— 指标作者在公式库中实现 INDICATOR 类，用 line / histogram / histogram2 / marker 四个函数输出指标图形；

- 第二类：画对象 —— 在 K 线图上手工/程序化绘制可编辑图形（趋势线、通道线、折线、斐波那契、文字标签、形状标记等），由 CandlePlot 统一管理（命名、拾取、拖拽、恢复）。

第一类：画指标

1.1 INDICATOR 类约定

指标文件位于 公式库/指标类/，图表通过 from <路径>.<名称> import INDICATOR 动态加载（加载前经沙箱预检）。

class INDICATOR:
    def __init__(self, MF, symbol, period):
        ...
    def OnCalculate(self, Data):
        ...   # Data 为行情 DataFrame，计算结果写入 indicator_bufferN

类属性约定（draw_object L4740 消费）：

1.2 样式元组与四个绘图函数

样式元组格式：[类型, 名称, 颜色, 宽度(大小), 附加(可选)]，类型 ∈ line / histogram / marker（histogram2 供程序直接调用，指标样式一般不用）。

指标作者不直接调用绘图函数——draw_object 读取样式元组后按类型分发调用以下函数（了解参数含义即可）：

① line —— 画折线（L4607）

cp.line(obj_ax, obj_price, obj_type, obj_name, obj_color, obj_width,
        obj_linetype='-', indicator_empty=None)

样式元组示例：['line', 'dif', 'white', 1]、['line', '量比', 'blue', 1, '-']

② histogram —— 正负双色柱（L4655，MACD 柱）

cp.histogram(obj_ax, obj_price, obj_type, obj_name, obj_mulcolor,
             obj_width, obj_linetype='-', indicator_empty=None)

• obj_mulcolor 传颜色列表可实现逐柱变色（红/绿交替）：

cmap = {True:'red', False:'lime'}; self.indicator3[2] = [cmap[v>0] for v in data]

• 柱宽自动按 K 线间距计算（约 1/4 间距，贴齐 K 线）

③ histogram2 —— 0 轴向上柱（L4649，从 0 到正数的柱形，负值部分压在 0 轴下）

cp.histogram2(obj_ax, obj_price, obj_mulcolor, obj_type, obj_name,
              obj_color, obj_width, obj_linetype='-', indicator_empty=None)

④ marker —— 标记点（L4694，信号箭头/圆点等）

cp.marker(obj_ax, obj_price, obj_type, obj_name, obj_color,
          obj_width, obj_marker, indicator_empty=None)

样式元组示例：['marker', '倍量信号', 'red', 20, 'o']

信号点做法：数据只在信号位置有值、其余为 NaN —— data['signal'] = data['量比'].where(cond)

1.3 完整示例

副图 + 水平线 + 多色柱（MACD.py 节选）

class INDICATOR:
    def __init__(self, MF, symbol, period):
        self.IndicatorName = 'MACD'
        self.indicator_chart_main = False          # 副图
        self.indicator_plots = 3
        self.indicator_level = [[0], ['#800000'], [':'], [0.8]]   # 0 轴虚线
        self.indicator1 = ['line', 'dif', 'white', 1]
        self.indicator2 = ['line', 'dea', 'gold', 1]
        self.indicator3 = ['histogram', 'macd', 'green', 0.2]
        self.indicator_buffer1 = []
        self.indicator_buffer2 = []
        self.indicator_buffer3 = []
        self.put = {'fastperiod': 12, 'slowperiod': 26, 'signalperiod': 9}

    def OnCalculate(self, Data):
        D = iMACD(Data, 12, 26, 9)
        self.indicator_buffer1 = D.dif.copy()
        self.indicator_buffer2 = D.dea.copy()
        self.indicator_buffer3 = D.macd.copy()
        cmap = {True: 'red', False: 'lime'}        # 逐柱变色
        self.indicator3[2] = [cmap[v > 0] for v in self.indicator_buffer3]

信号点 + 参数联动水平线（倍量检测.py 节选）

        self.indicator1 = ['line', '量比', 'blue', 1, '-']
        self.indicator2 = ['marker', '倍量信号', 'red', 20, 'o']
        self.indicator_level = [[self.put['量能倍数']], ['red'], ['--'], [1.0]]
        self.indicator_empty = 0

    def OnCalculate(self, Data):
        data['vol_ratio'] = data['volume'] / data['volume'].shift(1)
        cond = data['vol_ratio'] >= self.put['量能倍数']
        data['signal'] = data['vol_ratio'].where(cond)   # 非信号处 NaN
        self.indicator_buffer2 = data['signal'].values.tolist()

第二类：画对象（交互绘图工具）

2.1 工具类型与创建函数对照

画线工具由 UI 按钮经 set_drawTool(button, callback, itype) 激活（itype 即下表工具码），点击图表时按工具码分发到对应创建函数（L2454-2573）；cancle_drawTool() 取消。已画物件随图表保存并在重开时自动恢复（L5025-5035）。

2.2 统一参数说明

除 text（有 text/fontsize）、widget（有 style/size）、trendline（有 ray）外，其余函数参数一致：

2.3 坐标系 coord

程序化画对象时优先用 transData：x = cp.barShiftToTime(整数K线偏移)，y = 价格。

2.4 各工具要点

• 趋势线 CreatTrendLine：x=[x0,x1]、y=[y0,y1] 两点定线；ray=True 为射线（延长），False 为线段。
• 通道线 CreatChannel / 价格通道 CreatPriceChannel：x=[x0,x1,x0]、y=[y0,y0,y0+dy] 三点定义——前两点定基准线方向，第三点定通道宽度；两者仅默认线色与宽度差异。
• 手绘折线 CreatZig：逐点追加模式——首击创建，后续点击自动把新点接到折线尾部（L2528-2550），用于手工描画走势/ZigZag 结构。
• 斐波那契 CreatFibo：x=[x0,x1]、y=[y0,y1] 两点定区间，自动按经典比例（0/23.6/38.2/50/61.8/100…）画出回撤水平线组。
• 文字标签 CreatText：text 为内容，fontsize 默认 '18'；放置后可直接在图上编辑文字。
• 形状标记 CreatWidget：style 为 matplotlib 标记样式，size 为大小倍数（UI 点击放置时固定 size=2.0）。物件字典中 type='wing'（注意不是 'widget'），样式存 obj['style']、大小倍数存 obj['width']。

形状标记的 14 种形状与确定机制

UI 形状面板按固定顺序统一识别（与画线工具面板一致）：

画哪个形状由什么确定——三层机制，统一以 matplotlib 标记样式字符串为准：

1. 当前形状寄存器：cp.widgetstyle（默认 'o'，L145）。UI 形状面板点选后经 set_widgetstyle(style) 更新（同步 UI 变量），随后点击图表放置的就是该形状。
2. 交互放置：工具码 widget 激活后，点击图表 → CreatWidget(..., style=self.widgetstyle, size=2.0)（L2571）。
3. 程序化指定：调用 CreatWidget(..., style='D', size=2.0) 直接指定，不经寄存器。

识别已放置标记的形状：读物件字典 obj['style']（obj['type'] == 'wing'）。

2.5 物件管理

• 每个物件在 cp.objects[窗ID][名称] 存一份属性字典：type / name / color / linestyle / linewidth / xy / coord / explain / alpha / edit / select 及 matplotlib 对象句柄 object1 / object2（text 类多 text / fontsize，wing 类多 style / width，fibo 类多 fibo_value）。
• type 取值：hline / vline / trendline / channel / pricechannel / fibo / zig / text / wing（形状标记是 wing）。
• 选中/拖拽：点中物件进入编辑态（obj_handle['select']），控制点拖拽改坐标；-2 表示未选中。
• 恢复：切换品种/周期或重开软件时，按属性字典逐个调用创建函数重绘（x 重新经 barShiftToTime 映射到新 K 线轴）。
• 持久化：save_obj()（L4970）把物件属性（剔除运行时句柄 object1/object2/point）存为 ./config/<账号>/obt/<品种代码>（JSON，objects 字段）；换品种/重开经 download_objectList 读回重绘。

第三类：读取数据——程序如何取指标值与划线对象的值

以下读取方式全部对照源码核实（指标读取参照官方实现 show_ind_datas L3101；物件字段对照各创建函数的 obj 字典与持久化 save_obj L4970），代码可直接复制使用。

3.1 数据源总览

窗 ID：0 = 主图（K 线），1、2… = 副图（指标窗）。

3.2 读取指标值

指标数据在 INDICATOR 实例的 indicator_bufferN 上（list 或 pd.Series），按 K 线偏移取值。完整可用函数：

def get_indicator_value(cp, ind_name, where, ax_id=None):
    """取指标在某根K线的值。where 传 K线偏移(int) 或时间串(str)。
    返回 {输出名: 值}；找不到指标/越界返回 None"""
    shift = cp.barTimeToShift(where) if isinstance(where, str) else int(where)
    if shift is None or not (0 <= shift < len(cp.date_tickers)):
        return None
    axes_range = range(len(cp.funs)) if ax_id is None else [ax_id]
    for i in axes_range:
        for fun in cp.funs[i]:
            if fun.IndicatorName != ind_name:
                continue
            vals = {}
            for n in range(1, fun.indicator_plots + 1):
                label = getattr(fun, 'indicator%d' % n)[1]      # 输出名，如 'dif'
                data = getattr(fun, 'indicator_buffer%d' % n)
                if isinstance(data, list) and 0 <= shift < len(data):
                    vals[label] = data[shift]                   # list 直接下标
                elif isinstance(data, __import__('pandas').Series):
                    vals[label] = data.iloc[shift]              # Series 用 iloc
            return vals
    return None

# 用法：MACD 在最新一根K线的 dif/dea/macd
cp = self.main_frame.CP
vals = get_indicator_value(cp, 'MACD', len(cp.date_tickers) - 1)
# 或按时间：get_indicator_value(cp, 'MACD', '2026-09-29 10:30')
# vals -> {'dif': 0.123, 'dea': 0.098, 'macd': 0.05}

指标实例本身也可直接访问（改参数后触发重算等场景）：

for fun in cp.funs[0]:
    if fun.IndicatorName == 'MACD':
        dif_buf = fun.indicator_buffer1      # 整条 buffer（list/pd.Series）
        params = fun.put                     # 参数字典

3.3 读取划线对象：类型 → 点数 → 取值

所有对象的坐标统一存在 obj['xy'] = [[x…], [y…]]（x 为时间串，y 为价格）；辅助点数因类型而异。x 时间串换算 K 线偏移：cp.barTimeToShift(x_str)。

通用遍历函数（直接可用，一次取出全部对象的规范化数据）：

def get_objects_data(cp, ax_id=None):
    """读取图表上全部划线对象，返回规范化 list[dict]。
    每项含 name/type/points（[{time,shift,price}...]）+ 类型专属字段"""
    out = []
    axes_range = range(len(cp.objects)) if ax_id is None else [ax_id]
    for i in axes_range:
        for name, obj in cp.objects[i].items():
            item = {'ax': i, 'name': name, 'type': obj['type'],
                    'explain': obj.get('explain', ''), 'color': obj.get('color', '')}
            xs, ys = obj.get('xy', [[], []])
            item['points'] = []
            for x_str, y in zip(xs, ys):
                shift = cp.barTimeToShift(x_str) if isinstance(x_str, str) else x_str
                item['points'].append({'time': x_str, 'shift': shift, 'price': y})
            if obj['type'] == 'text':
                item['text'] = obj.get('text', '')           # 文字标签的内容
                item['fontsize'] = obj.get('fontsize', '')
            elif obj['type'] == 'wing':
                item['style'] = obj.get('style', 'o')        # 形状标记的形状
                item['width'] = obj.get('width', 1.0)
            elif obj['type'] == 'trendline':
                item['ray'] = obj.get('ray', True)
            elif obj['type'] == 'fibo':
                item['fibo_levels'] = fibo_prices(obj)       # 见 3.4
            out.append(item)
    return out

# 用法：列出主图全部对象
objs = get_objects_data(cp, ax_id=0)
for o in objs:
    print(o['name'], o['type'], o['points'])

3.4 斐波那契各档数值的读取

物件只存两个端点（obj['xy']）与比例表（obj['fibo_value']，缺省 10 档，见下）；每档水平线价格按源码公式（L2018）计算：

def fibo_prices(obj):
    """fibo 对象 -> {档位标签: 价格}。公式: yi = y1 - (y1 - y0) * ratio"""
    xs, ys = obj['xy']
    (x0, x1), (y0, y1) = xs, ys
    table = obj.get('fibo_value') or {'0.0': [0.0], '23.6': [0.236],
            '38.2': [0.382], '50.0': [0.5], '61.8': [0.618], '81.2': [0.812],
            '100.0': [1], '161.8': [1.618], '261.8': [2.618], '423.6': [4.236]}
    return {label: y1 - (y1 - y0) * ratio[0] for label, ratio in table.items()}

# 用法
objs = get_objects_data(cp, ax_id=0)
fibo = [o for o in objs if o['type'] == 'fibo'][0]
# fibo['fibo_levels'] -> {'0.0': 25.0, '23.6': 22.3, '38.2': 20.4, '50.0': 19.0, ...}

缺省 10 档比例（self.fibo_value L153，用户可在工具面板改，改后存于 obj['fibo_value']）：

0.0 / 23.6 / 38.2 / 50.0 / 61.8 / 81.2 / 100.0 / 161.8 / 261.8 / 423.6。

注意：fibo 两个端点的顺序即方向——y[1] 是“终点价”，比例从终点往起点方向回撤。

3.5 手绘折线（zig）取全部折点

zig = [o for o in get_objects_data(cp, 0) if o['type'] == 'zig'][0]
times  = [p['time']  for p in zig['points']]     # 折点时间串列表
shifts = [p['shift'] for p in zig['points']]     # 对应 K 线偏移（可用于取 OHLC）
prices = [p['price'] for p in zig['points']]     # 折点价格列表
# 配合行情取每段涨跌：cp.closes[p['shift']]

3.6 从持久化文件读（不打开图表时）

from base import readJson
data = readJson('./config/%s/obt/%s' % (useraccount, 品种代码))
# data['objects'] = [ {物件名: {type/xy/color/text/style/...}}, ... ]  按窗ID排列
# 与内存 cp.objects 的差别：无 object1/object2/point 运行时句柄，其余字段一致

附：程序化调用入口速查

cp_obj = self.main_frame.CP            # 当前 CandlePlot 实例

# 画对象（示例：在最新K线价格位置画一个红色圆点标记）
shift = cp_obj.barTimeToShift(cp_obj.barTime())
cp_obj.CreatWidget(cp_obj.axes[0], '', [cp_obj.barShiftToTime(shift)], [价格],
                   coord='transData', color='red', style='o', size=2.0)

# 画对象（示例：趋势线）
cp_obj.CreatTrendLine(cp_obj.axes[0], '', [t0, t1], [p0, p1],
                      coord='transData', color='gold', ray=False)

# 交互模式开关
cp_obj.set_drawTool(button, callback, 'trendline')   # 激活某画线工具
cp_obj.set_widgetstyle('s')                          # 切换形状标记样式
cp_obj.cancle_drawTool()                             # 取消画线模式

程序化监控策略开发指南

程序化监控是"每 N 秒扫描一次全市场/板块股票 → 命中信号弹窗播报"的盘中监控机制。

本文说明监控的轮询周期由什么控制、怎么设置，以及监控策略的编写规范。

一、监控是怎么跑起来的（工作机制）

点工具栏"执行量化"总开关后，软件的数据引擎进程会为每个程序化策略启动一条监控线程，流程如下：

启动程序 → 预备（预热监控品种的分钟K线内存表） → 运行（轮询循环）
运行轮询循环：
    1. 把监控范围内所有股票的 OnTime(code) 提交到线程池执行（每批 50 只）
    2. 一轮全部执行完 → 等待 Timer 秒
    3. 回到第 1 步，直到暂停/停止

策略需实现 OnTime(self, code) 方法，引擎会反复调用它；两次调用之间隔多久，由策略的 Timer 属性决定（见下节）。Timer 必须显式声明：策略不声明 Timer 时 OnTime 不生效（启动检测会打日志说明原因）。另外，策略若声明 TickSymbol（宿主品种），报价驱动的 OnTick 也会并行生效——完整的双入口机制见时间序列和指标访问。

二、监控周期设置 — self.Timer（重点）

概述

Timer 是策略类的一个普通属性，在策略的 __init__ 方法里自己定义，单位是秒。引擎启动监控时读取它，作为每轮全市场扫描之间的等待时间。不同策略可以设置不同的 Timer，互不影响。

定义方式

class STRATEGY:
    def __init__(self, MF):
        self.MF = MF
        self.strategy_name = '我的监控策略'
        self.put = {}
        # 设定监控轮询间隔（秒）
        self.Timer = 5          # 每 5 秒扫描一轮

提示：self.put 是策略的输入参数字典，配合 self.Enum 可以让参数在设置界面变成下拉菜单。完整规范见 输入参数与枚举下拉。

参数说明

取值建议

注意：轮询间隔 ≠ 信号计算频率。Timer 越小，OnTime 被调用的次数越多；如果每次都做真实计算，网络请求和 CPU 开销会成倍增加。建议配合第五节的"时间闸门"模式使用。

三、单只股票节流延迟 — put['监控延迟间隔(秒)']

与全局轮询周期是两个不同的东西：这是策略 OnTime 内部在处理完单只股票后的 time.sleep 延迟（防止连续请求过快、或命中信号后避免重复弹窗）。

def __init__(self, MF):
    self.put = {}
    self.put['监控延迟间隔(秒)'] = 2   # 每只股票处理完歇 2 秒再处理下一只

def OnTime(self, code):
    ...
    if 命中信号:
        self.MF.Alert('...出现进场点', programname=self.strategy_name, code=self.code)
        time.sleep(self.put['监控延迟间隔(秒)'])
        return(code)
    else:
        time.sleep(self.put['监控延迟间隔(秒)'])
        return(False)

该参数会出现在程序化窗口的参数面板里，用户可以在界面上修改，不需要改代码。

四、OnTime 方法的编写约定

def OnTime(self, code):
    # code 为带市场后缀的标准代码，如 '600000.SH'
    # 1. 判断是否到了计算时点（没到就直接返回 False，见第五节）
    # 2. 读取行情数据（用 api / MF.DC 统一接口）
    # 3. 计算信号
    if 命中信号:
        self.MF.Alert('提示文字', programname=self.strategy_name, code=self.code)
        return(code)      # 返回 code（非 False）→ 引擎判定为有信号
    else:
        return(False)     # 返回 False → 无信号

约定要点：

五、高级用法：策略内部时间闸门（timeList / calTime）

Timer 控制的是"多久看一眼"，如果策略只想在特定时点真正计算（例如每天 4 个整点），可以在 OnTime 里加时间闸门：高频轮询 + 低频计算，既有秒级反应速度，又几乎不产生无效计算。

def __init__(self, MF):
    ...
    self.Timer = 5                                        # 每 5 秒轮询一次（看表）
    self.timeList = ['9:30', '10:30', '13:00', '14:00']   # 真正计算的时点
    self.calTime = {}                                     # 每只股票的下一个计算时刻

def CT(self, iTime):
    """求 iTime 之后的下一个时间闸门时刻"""
    for x, t in enumerate(self.timeList):
        timeflag = datetime.datetime.strptime('%s %s' % (str(iTime.date()), t), '%Y-%m-%d %H:%M')
        if timeflag > iTime:
            break
        else:
            if x == len(self.timeList) - 1:
                timeflag = datetime.datetime.strptime('%s %s' % (str(iTime.date()), t), '%Y-%m-%d %H:%M') + datetime.timedelta(days=1)
            else:
                continue
    return(timeflag)

def OnTime(self, code):
    if code not in self.calTime.keys():
        self.calTime[code] = self.CT(datetime.datetime.now())
    # 读市场时钟
    _tickstr = ('%s %s' % (self.MF.CP.ticks['date'].get(), self.MF.CP.ticks['time'].get())).strip()
    try:
        self.ticktime = pd.to_datetime(_tickstr, format='%Y-%m-%d %H:%M:%S')
    except ValueError:
        return(False)                      # 报价未就绪，本次跳过
    if self.calTime[code] <= self.ticktime:   # 到达计算时点才做真实计算
        ...                                    # 读数据、算指标
        self.calTime[code] = self.CT(self.ticktime)   # 推进到下一个闸门
        if 命中信号:
            self.MF.Alert('...', programname=self.strategy_name, code=self.code)
            return(code)
        return(False)
    else:
        return(False)                      # 还没到时点，快速返回

完整可参考的实例见公式库：公式库/策略类/TakeOffStra.py、公式库/策略类/TakeOffStra2.py。

六、监控范围与运行状态

注意事项

• Timer 单位是秒，不要误写成分钟（Timer = 5 是 5 秒，不是 5 分钟）
• 轮询是在数据引擎独立进程中运行的，与主界面完全隔离，监控大量品种不会卡界面
• 一轮扫描必须全部执行完才开始计时等待 Timer 秒；如果监控范围很大、单轮耗时超过 Timer，实际周期以单轮耗时为准
• OnTime 中不要做耗时超过单轮周期的阻塞操作，否则信号会滞后
• 引擎批量提交每批 50 只，"停止程序"后队列中最多残留一批任务，属正常现象

输入参数与枚举下拉（put 与 Enum）

self.put 和 self.Enum 是软件所有公式体系统一的输入参数书写法——图表指标、交易策略（实盘 / 测试）、选股策略都使用同一套规范。

其中 Enum 是软件自创的参数下拉菜单机制：配上它，参数在设置界面从"手工输入框"变成"下拉菜单选择框"。

一、put：输入参数

self.put 是一个字典，存放公式对外暴露的可调参数。在 __init__ 里逐个赋默认值：

self.put = {}
self.put['N'] = 14                  # 数字参数
self.put['监控延迟间隔(秒)'] = 2     # 中文参数名（推荐，界面直接可读）
self.put['ma_method'] = 'EMA'       # 字符串参数

• 参数名可以用中文，用户在设置界面看到的名称就是键名。
• 公式内部任何地方用 self.put['参数名'] 取值。
• 用户在设置界面修改参数后，软件把新值写回 put 再触发重算。

二、Enum：下拉菜单参数（软件自创写法）

书写规则

self.Enum = {'交易方向':['双向','仅做多','仅做空']}
self.put['交易方向'] = '双向'

一个 Enum 字典可以写多对键值，为多个参数同时配置下拉菜单：

self.Enum = {'周期':['月线','周线','日线'], '均线类型':['SMA','EMA','SMMA']}
self.put['周期'] = '日线'
self.put['均线类型'] = 'SMA'

选项类型不限

列表里的可选项可以是字符串、布尔值或数字，界面按文本显示，选中后按原类型写回 put：

self.Enum = {'精选级别':[1,2,3],            # 数字选项
             '短期':[True,False],           # 布尔开关（选股公式常用惯例）
             '倍量阳':['所有倍量阳','首次倍量阳','不启用']}   # 字符串选项

支持范围

四处的参数设置界面统一识别 Enum：参数出现在 Enum 字典中就渲染为下拉框，否则是手工输入框。

三、策略内取值判断

以交易策略的"交易方向"为例（取自内置策略 DoubleMA）：

self.Enum = {'交易方向':['双向','仅做多','仅做空']}
self.put['交易方向'] = '双向'

# OnTime / 策略逻辑里：
if self.put['交易方向'] == '仅做多':
    ...   # 跳过做空分支

注意事项

• Enum 键名与 put 参数名必须逐字一致，写错一个字下拉就不会出现（该参数退化为普通输入框）。
• put 默认值必须是选项列表中的一项，否则设置界面无正确选中项。
• 选项值类型要一致（一个列表里不要混排字符串和数字），便于用户理解。
• 参数的合法性优先用枚举约束（用户只能选、不会输错）；无法枚举的数值（如均线周期）才用普通 put 参数加代码内自检。

交易函数（下单 / 平仓 / 持仓查询）

交易函数挂在策略的 self.order 对象上（orders 类），策略在 OnTick / OnTime 里调用它们完成下单、平仓、查询持仓。

当前为回测撮合：以策略触发当时那根 K 线的收盘价成交，资金按品种类型检查占用（股票全额、杠杆品种按保证金）。实盘对接层另行扩展。

职责边界：策略代码里的交易只对接回测撮合引擎。实盘下单不开放给策略代码——请使用软件界面的"交易助手"（人工确认后经券商通道发出）。在策略里 import tradeAPI 之类属于跨层调用，不允许、也会被沙箱拦截。

一、在策略中获得交易对象

from orders import orders

class STRATEGY:
    def __init__(self, cp=None):
        self.cp = cp
        self.order = orders(self)      # 交易对象：下单 / 平仓 / 持仓查询

常用属性：

二、函数一览

三、统一下单入口 send（推荐）

self.order.send(symbol=code, lot=100, orderType='buy',  price=price,
                stoploss=10.5, takeprofit=11.0, comment='金叉', magic=101)
self.order.send(symbol=code, lot=100, orderType='sell', price=price)

订单类型：

send 在回测时同步执行（买卖箭头画在触发信号的那根 K 线上，不会漂移）；程序化监控等异步场景投递队列由监听线程执行。

返回值语义：回测同步路径返回真实成交结果——资金不足等拒单返回 False，策略可据此决定下一步；监控异步路径返回 True 仅代表订单已入队，拒单发生在监听线程并打印日志。挂单类型返回 True 表示已登记挂单（非成交），触发发生在后续行情拍的撮合检查（见"六、挂单"）。

四、直接下单 buy / sell

buy / sell 是 send 的同步直通版本，参数一致。开仓时引擎自动完成：

1. 资金检查：按品种类型计算占用资金（股票 = 全额本金；杠杆品种 = 保证金），不足直接拒单并打印原因，返回 False；成功返回 True
2. 记入持仓列表 Orders，从 balance 扣除占用资金
3. 回测图表显示时在开仓 K 线上画买入（洋红）/ 卖出（绿色）箭头

资金占用规则（按品种代码自动识别类型）：

名义价值 = 价格 × 数量 × 合约乘数（股票乘数为 1）。

price = self.data.close.values[-1]
iLot  = int(self.order.balance / price * 0.5 / 100) * 100   # 半仓，取整百
if self.order.send(symbol=self.code, lot=iLot, orderType='buy', price=price) is False:
    print('开仓被拒绝（资金不足）')

五、平仓 close

profit = self.order.close(ticket=3, comment='死叉平仓')

• 以当前 K 线收盘价全额平仓（暂不支持部分平仓）
• 利润公式：买入 = 平仓市值 − 开仓名义；卖出 = 开仓名义 − 平仓市值
• 自动退回占用资金、结转利润到 balance，写入历史列表
• 图表上画平仓箭头和开仓→平仓虚线连线
• 返回该笔利润（数值）；订单号不存在打印原因并返回 None

配合 get_position 按魔术号批量平仓：

pos = self.order.get_position()
for _, row in pos[pos.magic == 101].iterrows():
    self.order.close(ticket=row['ticket'], comment='策略平仓')

六、挂单（触发 / 到期 / 撤单）

挂单是"未成交的订单"：登记后挂在引擎的挂单表中等待报价触及，不预占资金（触发时才做资金检查，不足则挂单作废）。

触发规则（q = 当前最新报价，p = 挂单价）：

撮合由引擎在每拍行情（回测 = 每根 K 线 / 分钟切片；监控 = 每次行情更新）自动执行，先于策略逻辑；触发后走 buy / sell 转正常持仓（挂单上设置的止损止盈随之生效）。期限到期自动撤单。

# 回测：跌破 9.5 买入，48 小时内未触发自动撤单
self.order.send(symbol=self.code, lot=100, orderType='buyLimit',
                price=9.5, expire_seconds=48*3600)

n = self.order.get_pending_count()          # 当前挂单笔数
pdTab = self.order.get_pending()            # 挂单列表（副本，可安全筛选）
if not pdTab.empty:
    self.order.cancel_order(int(pdTab.iloc[0]['ticket']))   # 撤第一笔
self.order.cancel_all(magic=7)              # 撤掉本策略批次全部挂单

• expire_seconds 从挂单时刻起按秒计时：回测用 K 线时间、监控用本机时间——挂单时刻与到期判定来自同一时间源做相对比较，无时区问题；0 = 不设期限
• 手动撤单 / 到期撤单 / 资金不足作废都记入 historys（reason 分别为 手动撤单 / 到期撤单 / 资金不足），用 get_history 可查挂单完整生命周期
• 回测撮合粒度 = 报价回放节奏（与止损止盈盯盘同源），非逐 tick

七、查询持仓 get_position

pos = self.order.get_position()          # 全部持仓
pos = self.order.get_position('600000.SH')   # 指定品种
if not pos.empty and self.sign == -1:
    self.order.close(ticket=int(pos.iloc[-1]['ticket']))

返回持仓 DataFrame 的副本（可安全筛选，不影响内部数据），列字段：

historys 另有：closetime / closeprice / reason / profit（单笔利润）/ profits（累计）。

八、查询历史 get_history 与笔数统计

his = self.order.get_history()                 # 全部历史（平仓记录 + 出入金记录）
his = self.order.get_history('600000.SH')      # 指定品种的历史
nPos = self.order.get_position_count()         # 当前持仓笔数
nHis = self.order.get_history_count()          # 历史记录笔数

• get_history 返回 historys 的副本（可安全筛选，不影响内部数据），字段与持仓表一致，另有 closetime / closeprice / reason（平仓原因）/ profit（单笔利润）/ profits（累计）
• 出入金记录也在历史列表里（品种列为空），按品种筛选时自然排除
• 笔数统计等价于对查询结果计数；symbol=None 统计全部

用历史记录做盈亏统计（lot 非空即成交记录，可排除出入金行）：

his = self.order.get_history()
deals = his[his.lot.notna()]
if not deals.empty:
    wins = deals[deals.profit > 0]
    print('累计盈亏 %.2f | 胜率 %.0f%% | 最大单笔亏损 %.2f' % (
        deals.profit.sum(), 100.0 * len(wins) / len(deals), deals.profit.min()))

九、出入金 writeIn

self.order.writeIn(money=1000000, comment='初始资金')      # 入金 100 万
self.order.writeIn(itype='出金', money=-50000)             # 出金 5 万

余额增减写入历史列表，用于初始化账户资金或在回测中模拟追加 / 提取资金。date 缺省取当天。

十、自动止损止盈（引擎功能）

send / buy / sell 里设置了 stoploss / takeprofit 后，回测引擎每根 K 线自动检查：盘中最低价触及止损价、或最高价触及止盈价即自动平仓（多空方向各自对应），策略不需要自己写盯盘逻辑；不设置则只按策略信号平仓。

完整示例：双均线开平仓

from Indicators import iMA
from orders import orders

class STRATEGY:
    def __init__(self, cp=None):
        self.cp = cp
        self.order = orders(self)
        self.strategy_name = '双均线策略'
        self.put = {'快线周期': 5, '慢线周期': 20, '仓位百分比': 0.5,
                    '止赢': 1, '止损': 0.5, '交易方向': '双向'}
        self.Enum = {'交易方向': ['双向', '仅做多', '仅做空']}

    def OnTick(self, code, data):
        self.data, self.code = data, code
        nF, nS = self.put['快线周期'], self.put['慢线周期']
        if data is None or len(data) < max(nF, nS) + 2:
            return False
        fast = iMA(data.close, nF, 0, 'EMA')
        slow = iMA(data.close, nS, 0, 'EMA')
        price = data.close.values[-1]
        golden = fast.values[-1] > slow.values[-1] and fast.values[-2] <= slow.values[-2]
        death  = fast.values[-1] < slow.values[-1] and fast.values[-2] >= slow.values[-2]
        if golden and self.put['交易方向'] != '仅做空':
            lot = int(self.order.balance / price * self.put['仓位百分比'] / 100) * 100
            self.order.send(symbol=code, lot=lot, orderType='buy', price=price,
                            stoploss=round(price - self.put['止损'], 2),
                            takeprofit=round(price + self.put['止赢'], 2),
                            comment='金叉', magic=1)
        if death:
            for _, row in self.order.get_position().iterrows():   # 反向信号全平
                self.order.close(ticket=int(row['ticket']), comment='死叉平仓')

注意事项

• 下单价格统一取当前 K 线收盘价 self.data.close.values[-1]，不要用实时 tick 价（回测按 K 线撮合）。
• 股票手数建议取整百（int(x/100)*100），避免成交数量不合法。
• 开仓前引擎做资金检查，拒单返回 False，策略可据此跳过后续逻辑。
• close 需要 ticket，平仓前先 get_position 查询；用 magic 区分不同策略批次可避免误平。
• 挂单不预占资金：触发时才做资金检查，不足则作废记历史（reason=资金不足）；市价单忽略 expire_seconds。
• 回测利润为纯价差：手续费 / 利息字段（commision / lixi）已保留但暂未计算，统计盈亏时不需要再扣成本。
• 拒单排查：开仓返回 False 时先看日志拒单原因（资金不足会打印需占用金额与可用余额），再确认代码后缀合法、手数取整百。
• close 的结果判断用 result is None——利润可能恰好为 0，不要用 if result: 真值判断。
• 监控场景 send 返回 True 只代表订单已投递队列；如需严格确认成交，用 get_position / get_history_count 复查。

股票函数

股票函数是吉宽量化中访问行情、下载数据、查询公司信息的核心 API。

软件所有涉及股票数据的功能（策略选股、回测引擎、K 线图表）都必须通过本模块的统一接口访问数据，不得绕过接口直接读文件或爬数据源。

统一接口层 —— 策略用户（写 EA / 量化程序 / 指标）唯一允许调用的数据接口层。

已实现的接口在策略代码中 import api 后调用（如 api.get_stock_kline(...)）。

子模块

接口一览（快速索引）

状态：已实现 = 代码已落地并测试通过（api.py）；预留 = 签名已固定、可安全调用（当前恒返回空，接入数据源后自动生效）。

读取（上层策略 / 回测 / 图表调用）

下载与管理（盘后下载窗口调用）

两套取数语义（均为正式接口）

旧接口（DC.get_market_data 等）保证不改名、不删除，现有指标文件不受影响；

新代码请统一使用 api.py 的标准接口。

股票函数 · 基本数据

本章接口由 api.py 提供（策略用户统一 API 第一层）。

已实现的接口用完整文档说明；尚未实现的接口在文末"规划中"一节列出，

待实现后逐个补充完整文档（一项一项讨论通过后发布）。

get_stock_namelist — 获取全市场股票列表（已实现）

概述

返回当前股票列表（A股：主板、创业板、科创板、北交所），含代码、名称、市场、拼音。

用于初始化股票池、遍历全市场、按市场筛选。

列表由软件的股票列表提供者进程从交易所官网（SSE / SZSE / BSE）定时抓取，

存入本地 stock.db 数据库的 stock_name_tb 表；本接口从该表读取，不做网络请求。

软件启动后约 1 分钟内列表会自动刷新并缓存到本地。

如果某只股票不在返回结果里，通常意味着该股票已停牌超过 30 天、退市或已不在当前交易所挂牌。

函数签名

get_stock_namelist(market='all')

参数

返回值

list[dict] — 每一项包含：

示例代码

import api

# 取全部
all_stocks = api.get_stock_namelist()
print('总数：', len(all_stocks))

# 筛选上交所
sh = api.get_stock_namelist(market='SH')
print('上交所：', len(sh), '只')

# 遍历前 3 只
for s in all_stocks[:3]:
    print(s['ts_code'], s['name'], s['exchange'])

输出结果（实测）

总数： 4947
上交所： 1702 只
600000.SH 浦发银行 SH
000001.SZ 平安银行 SZ
000002.SZ 万科A SZ

注意事项

• 首次安装软件后列表可能只有默认指数条目（上证指数等），提供者进程抓取完成后自动补全。
• 本接口返回的是股票列表；指数列表（用于取指数 K 线）不在此接口中，

指数 K 线直接用 get_stock_kline('000001.SH') 即可（自动识别为指数）。

get_stock_basic — 查询单只股票基本信息（已实现）

概述

返回一只股票的上市日期、行业分类、所在地区等静态信息。

数据来自本地股票列表库（与 get_stock_namelist 同源），不做网络请求。

常与 get_stock_namelist 搭配：先用列表拿到代码池，再逐只查基本信息做过滤

（如剔除银行股、只做某地区、剔除次新股）。

函数签名

get_stock_basic(symbol)

参数

返回值

dict — 字段：

查无此股返回空 dict {}。

示例代码

import api

info = api.get_stock_basic('000001.SZ')
print(info['name'], '上市于', info['list_date'], '行业:', info['industry'])

# 过滤股票池：只要 2020 年以前上市的老股票
pool = api.get_stock_namelist()
old = [s['ts_code'] for s in pool
       if api.get_stock_basic(s['ts_code']).get('list_date', '') < '20200101']
print('上市满 6 年的老股票：', len(old), '只')

输出结果（实测）

平安银行 上市于 19910403 行业: J 金融业

注意事项

• 首次安装软件、列表尚未抓全时，部分字段（list_date / area / industry）可能为空字符串。
• 指数（如 000001.SH）不在股票列表库中，本接口查询返回 {}；指数请直接用 get_stock_kline。

get_stock_calendar — 查询交易日历（已实现）

概述

给定起止日期，返回其中所有交易日（自动剔除周末、法定节假日、交易所特殊休市日）。

用于回测引擎校验日期、盘后下载 diff、计算"最近 N 个交易日"等场景。

日历从本地指数 K 线推导（上证指数每个交易日都有一条 K 线），与回测、下载使用的是同一套真实交易日。

函数签名

get_stock_calendar(exchange='SSE', start_date=None, end_date=None)

参数

返回值

list[str] — 升序排列的交易日期字符串；本地无指数数据返回空列表

示例代码

import api

# 2024 年所有交易日
cals = api.get_stock_calendar(start_date='20240101', end_date='20241231')
print('2024 年交易日总数：', len(cals))
print('元旦后第一个交易日：', cals[0])

# 找最近 5 个交易日
cals = api.get_stock_calendar()
print('最近 5 个交易日：', cals[-5:])

# 回测前校验：本地数据缺失哪些交易日
cals = api.get_stock_calendar(start_date='20250101', end_date='20260918')
df = api.get_stock_kline('000001.SZ', 'D1', start_date='20250101', end_date='20260918')
missing = [d for d in cals if d not in set(df['trade_date'])]
print('本地数据缺失交易日数：', len(missing))

输出结果（实测）

2024 年交易日总数： 242
元旦后第一个交易日： 20240102

get_stock_ipo — 新股上市列表（已实现）

概述

返回指定时间段内 IPO 上市的股票（按上市日期降序，最新的在前）。

策略选股可以利用这个接口剔除"刚上市 60 天"的次新股。

数据来自本地股票列表库的 list_date 字段（与 get_stock_namelist 同源），不做网络请求。

函数签名

get_stock_ipo(start_date=None, end_date=None)

参数

返回值

list[dict] — 每项字段：ts_code / name / list_date / market / exchange（list_date 无记录的股票自动排除）

示例代码

import api
import datetime

# 找出最近 60 天新上市的股票
end = datetime.datetime.now().strftime('%Y%m%d')
start = (datetime.datetime.now() - datetime.timedelta(days=60)).strftime('%Y%m%d')
ipos = api.get_stock_ipo(start, end)
print('近 60 天上市', len(ipos), '只')
recent = {s['ts_code'] for s in ipos}

# 选股时剔除次新股
pool = api.get_stock_namelist()
old_pool = [s for s in pool if s['ts_code'] not in recent]

输出结果（实测）

近 60 天上市 30 只
最新: 601091.SH C沈鼓 20260917

get_stock_namechange — 曾用名历史（预留接口，已实现）

概述

返回一只股票的历史名称变更（ST 警示、摘帽、借壳改名等），用于回测时正确识别

同一股票代码在不同时期的真实身份。

当前版本为预留接口：软件暂未接入曾用名数据源（需交易所/Tushare 付费接口），

本函数恒返回空列表 []。接口签名已固定，将来接入数据后返回值自动生效；

现在就可以在策略中安全调用并按空列表处理。

函数签名

get_stock_namechange(symbol)

参数

返回值

list[dict] — 当前版本恒为 []。将来每项将包含：begin_date / end_date / name

get_stock_hs_component — 沪深股通成分股（预留接口，已实现）

概述

返回某一日期的沪深港通（北向 / 南向）成分股列表。

当前版本为预留接口：软件暂未接入沪深港通成分股数据源，

本函数恒返回空列表 []。接口签名已固定，将来接入数据后返回值自动生效。

函数签名

get_stock_hs_component(date=None, side='north')

参数

返回值

list[dict] — 当前版本恒为 []。将来每项将包含：ts_code / name / market

股票函数 · 行情数据读取

本章接口由 api.py 提供（策略用户统一 API 第一层），已实现并通过实测验证。

当前版本在策略/指标代码中通过 import api 调用；将来沙箱运行环境就绪后，

这些函数会自动注入策略命名空间，无需 import 直接调用。

get_stock_kline — 获取 K 线历史数据（统一接口）

概述

这是写策略 / EA / 指标时获取历史 K 线的唯一推荐入口。

• range 型语义：按日期区间取数，最适合选股、回测、指标计算。
• 数据从本地读取（data/ 目录下的 parquet 文件），不做网络请求，速度快、可批量遍历全市场。
• 本地没有数据时返回空 DataFrame（不会自动联网下载），请先用软件的盘后下载窗口把数据下载到本地。

说明：软件内部（K 线图表、回测引擎）使用的取数接口是 DC.get_market_data

（count 型，取最近 N 根）。两套接口并存、都是正式接口；**新写的策略代码请统一使用

本接口**，旧接口我们保证不改名、不删除，现有指标文件不受影响。

函数签名

get_stock_kline(symbol, period='D1', market='cn-stock', asset_class=None,
                start_date=None, end_date=None, columns=None, strategy_cols=False)

参数

asset_class 自动识别规则（不用关心数据存在哪个目录）：

• 000xxx.SH（如 000001.SH 上证指数）、399xxx.SZ、899xxx.BJ、.CI 后缀 → 指数，读 data/cn-index/
• 其余代码 → 股票，读 data/cn-stock/

返回值

pd.DataFrame — 按 trade_date 升序，索引已 reset。

标准基本列（8 列，本版本承诺）：

• 无数据返回空 DataFrame（不是 None），调用前用 df.empty 判断。
• 返回的 DataFrame 可能包含内部预留列（当前版本值为空），属于内部扩展结构，

请不要依赖文档未列出的列，只用上面 8 个基本列。

示例代码

import api
import datetime

# 1) 日线取最近数据（自动识别股票）
df = api.get_stock_kline('000001.SZ', 'D1', start_date='20250101')
print('行数：', len(df))
print('最新交易日：', df['trade_date'].iloc[-1])

# 2) 策略选股场景：只取 OHLCV（列裁剪加速）
df = api.get_stock_kline('000001.SZ', 'D1', start_date='20250101', strategy_cols=True)

# 3) 取指数（自动识别为指数，读指数库）
df_idx = api.get_stock_kline('000001.SH', 'D1', start_date='20250101')   # 上证指数

# 4) 不带后缀，自动补全
df = api.get_stock_kline('600000', 'D1', start_date='20250101')          # → 600000.SH

# 5) 均线金叉示例
df = api.get_stock_kline('000001.SZ', 'D1', start_date='20240101', strategy_cols=True)
df['MA5']  = df['close'].rolling(5).mean()
df['MA20'] = df['close'].rolling(20).mean()
cross = df[(df['MA5'].shift(1) < df['MA20'].shift(1)) & (df['MA5'] > df['MA20'])]
print('金叉日期：', cross['trade_date'].tolist())

# 6) 遍历股票池
pool = api.get_stock_namelist()[:10]
for s in pool:
    df = api.get_stock_kline(s['ts_code'], 'D1', start_date='20250101', strategy_cols=True)
    if df.empty:
        print(s['ts_code'], '本地无数据，请先在盘后下载窗口下载')

输出结果（实测）

行数： 417
最新交易日： 20260918
金叉日期： ['20240206', '20240819', '20250711']

注意事项

• 数据完整性：本地没有的数据不会自动下载。判断本地数据够不够，用 check_data_coverage

接口（已提供，api.check_data_coverage(symbol, period)）；也可在盘后下载窗口查看覆盖情况。

• 周线 / 月线 / 分钟线：需要先用下载窗口下载过对应周期才有数据；从未下载过的周期返回空 DataFrame。
• 复权口径：close 为前复权价格。本版本不提供复权因子列，如需不复权价格请关注后续版本。
• 性能提示：strategy_cols=True 或显式 columns=[...] 读取更快；批量遍历全市场（约 4900 只）

串行调用本接口即可，单只平均约 0.01 秒。

• 新旧接口关系：DC.get_market_data(code, period, count=N) 是软件内部使用的 count 型接口，

仍然有效、不会删除；策略新代码请使用本接口。

取某一根 K 线的两种标准方法（按时间 / 按根数）

概述

策略里经常要拿某一根 K 线的时间 + 开高低收 + 成交量，两种定位方式：

系统不设专门的单根取数函数——两个正式接口加一行标准写法即可覆盖，避免接口冗余。

方法一：按时间取一根

get_stock_kline 是区间取数，把 start_date 和 end_date 写成同一个时刻就精确取一根；分钟线时刻精确到分钟 'YYYYMMDDHHmm'。

import api

# 日线：取 2026-09-24 这一天的开高低收 + 成交量
d = api.get_stock_kline('000001.SZ', 'D1',
                        start_date='20260924', end_date='20260924')
if not d.empty:
    print(d[['trade_date', 'open', 'high', 'low', 'close', 'vol']].iloc[0])

# 分钟线：取 09:30 开盘的那根 M30（T 必须写 bar 的开盘时刻，系统口径）
m = api.get_stock_kline('000001.SZ', 'M30',
                        start_date='202609240930', end_date='202609240930')

该时刻没有 K 线（停牌、非交易时段、本地未下载）返回空 DataFrame，取值前必须判 empty。

方法二：按根数（序号）取一根

get_stock_kline 返回按时间升序的 DataFrame，索引已 reset，用 iloc 按位置定位：

import api

df = api.get_stock_kline('000001.SZ', 'D1', start_date='20250101', strategy_cols=True)

bar   = df.iloc[-1]      # 最新一根：时间 + 开高低收 + 量 一整行
bar3  = df.iloc[-3]      # 倒数第 3 根
first = df.iloc[0]       # 顺数第 1 根

bar['trade_date'], bar['open'], bar['close'], bar['vol']   # 时间 / 报价 / 成交量

count 型等价写法（软件内部同款，取最近 N 根里的最后一根）：

D = self.MF.DC.get_market_data('000001.SZ', 'D1', 1)   # 只取最近 1 根
bar = D.iloc[-1]

MT4 shift 习惯对照：shift=0（当前根）→ iloc[-1]；shift=1（上一根）→ iloc[-2]；以此类推。回测策略内判断金叉/死叉用的就是这套惯例（见时间序列和指标访问）。

注意事项

• 按时间取分钟线必须给 bar 的开盘时刻（系统统一口径是 bar 开盘时间标签）：给成结束时刻不会报错，而是错拿另一根或查不到——例如 M5 里 0935 是 [09:35,09:40) 那根的开盘标签，你以为取的是 [09:30,09:35)，实际拿到的是下一根。口径详见格式标准第 0 节
• 图表端是另一套横轴：画线对象定位用 self.MF.CP.barTimeToShift(...) / barShiftToTime(...)（见图表绘图API），拿到的是图表偏移值，不要与数据端 iloc 位置混用
• 回测策略内不要用这两个方法反复取"当前根"——回测引擎每根 K 线已把数据窗口 data 递给 OnTick，直接 data.close.values[-1] 最快且无未来函数风险

get_realtime_quote — 获取实时行情快照

概述

返回某只股票当前时刻的实时行情快照（现价、开高低、涨跌幅、五档买卖盘等）。

数据来自软件实时数据进程写入共享内存的全市场快照（每约 6 秒从数据源刷新一轮），

本函数不做网络请求，可在策略中高频调用（取到的是最近一次刷新值）。

适用场景：盘中实时监控、实时信号触发。

不适用于回测——回测中的"当前价"请使用 K 线数据。

函数签名

get_realtime_quote(symbol, depth=5)

参数

返回值

dict — 字段：

• 股票不在快照中、或实时数据尚未就绪时返回空 dict {}。
• 实时行情上下文未绑定时（非正常软件环境下直接调用）抛 RuntimeError。

示例代码

import api

q = api.get_realtime_quote('000001.SZ')
if q:
    print(q['name'], q['price'], q['pct_chg'], '%')
    print('买一', q['bids'][0], q['bid_vols'][0], '卖一', q['asks'][0], q['ask_vols'][0])
else:
    print('实时数据尚未就绪（可能未开盘或数据进程未启动）')

输出结果（实测）

平安银行 11.85 1.28 %
买一 11.84 128.0 卖一 11.85 1371.0

注意事项

• 全市场快照约 6 秒刷新一轮，相邻两次调用可能返回相同值；本接口无反爬限制，可放心高频调用。
• 五档盘口最多 5 档，depth>5 按 5 处理。
• 集合竞价 / 停牌时段部分字段可能为空值（None）。

股票函数 · 格式标准

这是软件内部的权威契约 —— 任何数据源适配器、任何新接入的数据源、任何将来加的存储后端都必须遵守。

上层代码只调用统一接口（get_stock_kline / download_daily），绝不直接触碰底层文件 / 原始数据源。

0. 周期标准（系统统一表述）

系统内部所有周期一律使用以下标准表述——不管外部数据源的周期怎么写（1min / day / MIN_5 / 240……），进系统时必须先转换成标准周期；所有接口出口（get_market_data / get_stock_kline / 图表 / 策略）也只认标准周期。

标签口径：bar 开盘时间（系统统一约定）

每根 K 线的时间标签 = bar 的开盘时间（与外盘 / MT4 惯例一致），全软件所有周期统一：

• '0930' 标签的 M5 bar 覆盖 [09:30, 09:35)；周/月线标签 = 周期内首个交易日；日线为日期本身
• 国内数据源的分钟 bar 原始以结束时间为标签（'0935' 覆盖 09:30-09:35，通达信约定）——进系统前必须转成开盘标签，转换规则：结束时刻减去一个 bar 时长（A 股 H1 实测 pytdxdata：bar 从 09:30 起对齐的均匀 60 分钟，结束标签 1030/1130/1400/1500 各减 60 分钟 = 开盘 0930/1030/1300/1400，固定减法全部正确；跨日由 datetime 运算自然处理。若未来某数据源的 H1 对齐方式不同，由该数据源适配器先归一为本口径，全局转换器不做特判）
• 统一转换函数：bar_end_to_open(time_str, period) 单值 / bar_end_to_open_series(s, period) 整列 / bar_minutes(period) 周期分钟数（用法见日期和时间）
• 落盘链路：saveSymbolData(..., bar_labels='end')（默认）在存储唯一入口自动转换；实时合成等已统一口径的链路传 bar_labels='open' 原样落盘，防止二次平移
• 取数链路：get_market_data 等取数出口对网络源数据自动转换（本地存储源已是开盘口径，跳过）

入口转换（数据源适配层）：每个数据源用一张"标准周期 → 该源原生周期"的映射表（如 tushare 源 {'M1':'1min',...,'D1':'D','W1':'W','MN':'M'}、通达信源 {'M1':KlinePeriod.MIN_1,...}、腾讯源 {'D1':'day',...}）。映射的键必须写标准周期，源原生格式只出现在值里。接新数据源同理（见第 5 节）。

持久层 freq 与标准周期的关系：本地 tushare db 的 freq 参数（'1min'/'5min'/'D'/'W'/'M'）是存储实现格式，不是系统标准周期——只允许出现在"标准周期 → freq"映射的值里，任何接口参数、列名、判断分支都不得直接使用。

存储目录说明：data/<market>/<period>/ 的周期目录名就是标准周期（M1/M5/M15/M30/H1/D1/W1/MN，见第 1 节），代码里的周期键与目录名一致。

禁止：用 'M1' 表示月线（标准里 M1 = 1 分钟，月线是 MN）；新代码引入 1min/5min/60min 等外部写法。

1. 存储目录

data/
  <market>/              # cn-stock / us-stock / cn-future / fx
    <period>/            # 标准周期目录名 D1/W1/MN/M1/M5/M15/M30/H1
      <symbol>.parquet   # 000001.SZ.parquet / IF2501.parquet / EURUSD.parquet

例子：

data/cn-stock/D1/000001.SZ.parquet
data/cn-stock/D1/600000.SH.parquet
data/cn-stock/1min/000001.SZ.parquet
data/us-stock/D1/AAPL.parquet
data/cn-future/D1/IF2501.parquet
data/fx/1min/EURUSD.parquet

2. parquet 文件规范

• 压缩：snappy（默认参数，兼顾解压速度和压缩率）
• 按股票分片：一只股票一个文件（不是每天一个 / 每月一个 / 按行情一个）
• 增量写入：下载时先读已有文件 → 新数据按 trade_date 合并去重 → 整文件覆盖写回

（不做 append / replace partition — parquet 不原生支持行级修改）

3. 日线标准列（列顺序 = parquet schema）

必填列 7 个（前 7 行） 是 strategy_cols=True 要读的 OHLCV 子集。

非必填列留 NaN 即可，parquet 对空列几乎不占空间。

4. 分钟线标准列

注意：分钟线的列名仍然是 trade_date（与日线同名），区别只在取值格式：日线 'YYYYMMDD'，分钟线 'YYYYMMDDHHMM'（含时分，bar 开盘时间）。

get_stock_kline(..., period='1min') 返回的列名是 trade_date。

5. 数据源适配器契约（接新数据源必读）

必须实现

class MyNewSource(DataSourceBase):
    """我的新数据源 —— 例如 akshare / baostock / 商业 API"""
    
    def __init__(self):
        super().__init__(name='我的数据源', source_id=99)
    
    def is_available(self) -> bool:
        """轻量探测：能不能拉到一条数据"""
        return True
    
    def get_data_by_range(self, code, period='D1',
                          start_date=None, end_date=None,
                          market_type='STOCK') -> pd.DataFrame:
        """
        从我的数据源拉一只股票 [start_date, end_date] 区间的数据。
        
        返回 DataFrame 的列必须对齐"格式标准"的日线 / 分钟线规范：
          - 日线：trade_date, ts_code, open, high, low, close, vol, ...
          - 分钟线：trade_date, ts_code, open, high, low, close, vol, ...
        - trade_date 必须是 string：日线 'YYYYMMDD'，分钟线 'YYYYMMDDHHMM'
          （12 位紧凑格式，不带横杠 / 冒号 / 秒；时间标签口径见下方"时间标签口径"小节）
        - open/high/low/close/vol 必须是 float64
        - 列顺序不重要（pandas 会按列名对齐），但列名必须精确匹配
        - 按 trade_date 升序
        - 不能有空行（全零 OHLC 行要过滤掉）
        
        如果一次请求拿不全，自己在函数内实现分段 / 分页逻辑，
        返回给协调器的时候已经是完整的区间数据。
        
        反爬限速、重试、HTTP 头伪装 —— 全部由这个函数自己处理。
        """
        ...

绝对不能做

• 不要在这个函数里碰 data_coverage_tb —— symbol_io.saveSymbolData 会自动 upsert
• 不要自己写 parquet / CSV —— 返回 DataFrame，协调器会调用 saveSymbolData
• 不要自己起线程池 —— 外部协调器串行调用你的 get_data_by_range
• 不要 print 大量调试日志 —— 最多 print 最终汇总

数据源输出对齐代码示例

# 假设我用 baostock 拉到了原始 K 线
raw_rows = baostock.query_k_data(code='sh.600000', start_date='2024-01-01', ...)

# 必须转换成标准列
records = []
for row in raw_rows:
    records.append({
        'trade_date': row.date.replace('-', ''),  # '20240102'
        'open':      float(row.open),
        'high':      float(row.high),
        'low':       float(row.low),
        'close':     float(row.close),
        'vol':       float(row.volume),
        # 以下是非必填，留 None / NaN 即可
        'amount':     None,
        'pre_close':  None,
        'change':     None,
        'pct_chg':    None,
        'adj_factor': float(row.adjfactor) if row.adjfactor else None,
    })

df = pd.DataFrame(records)
df.drop_duplicates('trade_date', keep='last', inplace=True)
df.sort_values('trade_date', inplace=True)
return df.reset_index(drop=True)

时间标签口径（自接历史数据必读，防接错）

系统统一口径 = bar 开盘时间标签（与外盘 / MT4 惯例一致）：'0930' 标签的 M5 bar 覆盖 09:30–09:35，即标签时刻是这根 bar 的开始，不是结束。日线是日期本身，无开盘收盘之分；周 / 月线标签 = 周期内首个交易日。口径总说明见第 0 节"标签口径"。

你自己接入历史数据时，trade_date 的时间标签有两种合法口径，必须在落盘时声明清楚，否则数据整体错位一个周期：

接错的表现（自检方法——接完用 get_stock_kline 取头两根看标签；注意要用盘后完整下载的数据判断，实时监控固化的数据可能缺早盘头几根（监控启动晚），首根晚于 0930 不一定是接错）：

• 忘了声明 bar_labels='open'（开盘口径的数据走了默认 end 转换）→ 二次平移，每根 bar 整体前移一个周期：M5 第一根本该是 0930 变成了 0925
• 源是结束标签但自己手工转错 / 绕过 saveSymbolData 直接写文件 → 标签残留结束口径：M1 第一根是 0931、M5 第一根是 0935（正确应为 0930）
• 两种错位都会导致按时间取数错拿相邻 bar 或查不到、与本地已有数据拼接后同一天出现两根相邻 bar、指标整体漂移一根

接入后如何生效

1. 写好 MyNewSource 放在 DataSources/my_source.py
2. 在 list_registered_sources() 的注册表里登记
3. 把盘后下载窗口的"数据源"配置改成 ['my_source']

（UI 还没做这个配置入口，当前默认腾讯财经；后续再加下拉框）

4. 软件主体一行不用改 —— DailyDownloadCoordinator.run() 只认 DataSourceBase 抽象接口

6. data_coverage_tb 表结构

CREATE TABLE IF NOT EXISTS data_coverage_tb (
    market      TEXT NOT NULL,        -- cn-stock / cn-future / fx
    asset_class TEXT NOT NULL,        -- stock / future / forex
    period      TEXT NOT NULL,        -- D1 / 1min / 60min
    symbol      TEXT NOT NULL,        -- 000001.SZ / IF2501 / EURUSD
    first_time  TEXT NOT NULL,        -- 本地最早日期 YYYYMMDD
    last_time   TEXT NOT NULL,        -- 本地最新日期 YYYYMMDD
    rows        INTEGER NOT NULL,     -- 总行数
    source_path TEXT,                 -- parquet 文件路径
    file_fmt    TEXT,                 -- 'parquet'
    updated_at  TEXT,                 -- 最近一次更新时间戳
    PRIMARY KEY (market, asset_class, period, symbol)
);

写入由 symbol_io.saveSymbolData 自动 upsert，上层代码不要手动改这张表。

7. 与老格式的关系

软件早期版本用 ./history/stk/D1/<code>.csv 存日线。

新代码不再依赖这个目录（已切 parquet + data/cn-stock/D1/）。

老目录暂时保留作归档，将来可以手动删除。

任何新写的代码绝对不要去读老目录的 CSV —— 必须走 get_stock_kline 或 getSymbolData。

股票函数 · 盘后下载与数据管理

本章接口由 api.py 提供（策略用户统一 API 第一层）。

这组函数负责把原始行情数据拉到本地，供 get_stock_kline 读取。

其中 download_daily 是长耗时操作——更推荐直接使用软件的盘后下载窗口

（内部就是同一套下载引擎），本接口主要供脚本化批量更新、自动化维护使用。

download_daily — 批量下载日线数据（已实现）

概述

从数据源批量拉取日线数据，写入本地 data/cn-stock/D1/*.parquet，

并自动更新数据覆盖清单（data_coverage_tb）。

支持三种下载模式：

1. 全部历史：从每只股票上市日拉到今天（耗时取决于数据源：通达信批量源约 15~30 分钟，网页单只源数小时，建议夜间静默跑一次）
2. 三年内补全：起止日期自动推导（今天-1095天 ~ 今天），已覆盖的品种自动跳过（日常盘后推荐）
3. 指定区间：手动输入起止日期，常用于回测前补一段历史

下载方式由当前所选数据源自己决定（有批量接口的并发批量下，只有单只接口的逐只下），

详见 ../数据源接入/盘后下载接入规范.md。

函数签名

download_daily(start_date=None, end_date=None, mode='all', sources=None,
               progress_callback=None, cancel_check=None,
               period='D1', market='cn-stock', asset_class='stock')

参数

返回值

dict — 下载摘要：

{
  'success': 2891,
  'fail': 2,
  'elapsed': 3862.5,     # 秒
  'date_range': ('20230917', '20260916'),
  'fail_symbols': ['688123.SH', '301456.SZ']   # 具体哪几只失败
}

示例代码

import api
import threading

# 1) 日常盘后：三年内补全（后台线程跑，不阻塞脚本）
def job():
    result = api.download_daily(mode='diff')
    print('下载完成：成功', result['success'], '失败', result['fail'])

threading.Thread(target=job, daemon=True).start()

# 2) 回测前补一段历史 + 进度显示 + 可取消
result = api.download_daily(start_date='20200101', end_date='20221231', mode='all',
                            progress_callback=lambda i, t, m: print(f'[{i}/{t}] {m}'))
if result['fail_symbols']:
    print('失败的品种：', result['fail_symbols'])

输出结果（实测，cancel 立即取消的链路验证）

返回: {'success': 0, 'fail': 0, 'elapsed': 0.02, 'date_range': ('19000101', '20260920'), 'fail_symbols': []}
进度回调首条: (0, 4947, '准备下载 4947 个品种 （all 19000101~20260920） [跳过已有]')

注意事项

• 阻塞：本函数在当前线程串行下载，脚本里请自行放独立线程；UI 场景直接用盘后下载窗口。
• 下载节奏由数据源自定：网页单只源内部自带防反爬间隔；批量源（如通达信）内部并发拉取，耗时大幅缩短。
• 增量安全：下载数据与本地已有文件按 trade_date 合并去重后落盘，不会重复、不会覆盖掉本地新数据。
• 已完全覆盖所选区间的品种自动跳过（diff 模式和 all 模式都有此优化）。

check_data_coverage — 查询本地数据覆盖范围（已实现）

概述

给定一个 symbol + 周期，返回本地数据的覆盖区间和总行数。

策略选股、回测启动前、批量取数前都应该先调这个函数确认"本地数据够不够"。

函数签名

check_data_coverage(symbol, period='D1', market='cn-stock', asset_class=None)

参数

返回值

dict：

{
  'symbol':        '000001.SZ',
  'has_data':      True,
  'first_time':    '20230918',   # 本地最早日期
  'last_time':     '20260918',   # 本地最新日期
  'rows':          728,          # 总行数
  'file_path':     'data/cn-stock/D1/000001.SZ.parquet',
  'file_fmt':      'parquet',
  'coverage_ok':   True,         # last_time >= 全市场覆盖众数 → True
  'latest_common': '20260918'    # 全市场大多数品种覆盖到的最新日期
}

无记录品种返回 has_data=False，其余字段为 None/False。

symbol='_ALL_' 时只返回 symbol / has_data / latest_common 三个字段。

示例代码

import api

cov = api.check_data_coverage('000001.SZ', 'D1')
print('覆盖 OK：', cov['coverage_ok'], '最新：', cov['last_time'])

# 全市场众数（最有用的一个数——判断本地数据整体是否跟上了）
g = api.check_data_coverage('_ALL_', 'D1')
print('全市场日线最新覆盖：', g['latest_common'])

# 批量检查股票池
pool = api.get_stock_namelist()[:20]
notok = [s['ts_code'] for s in pool if not api.check_data_coverage(s['ts_code'])['coverage_ok']]
print('20 只里覆盖不足的：', notok)

输出结果（实测）

覆盖 OK： True 最新： 20260918
全市场日线最新覆盖： 20260918

注意事项

• coverage_ok 的标准是与全市场覆盖众数比较（与选股窗口"开始选股"的拦截判断同一标准），

停牌股票不会因此被误判为覆盖不足。

list_registered_sources — 当前接入的数据源清单（已实现）

概述

返回软件当前已经接入的数据源清单（名称 / ID / 支持周期 / 是否可用）。

将来新接数据源后会出现在这个列表里。

函数签名

list_registered_sources()

返回值

list[dict] — 每项字段：name / source_id / periods / available / note

示例代码

import api
for s in api.list_registered_sources():
    print(s['name'], s['available'], s['periods'])

输出结果（实测）

本地数据 id=2 available=True periods=['D1','W1','M1','1min','5min','15min','30min','60min']
腾讯财经 id=3 available=True periods=['D1','W1','MN','M5','M15','M30','H1']
新浪财经 id=0 available=True periods=['D1','tick']
Tushare 数据 id=1 available=False（需付费 token，未配置时不可用）
腾讯财经日线(盘后下载) id=None available=True periods=['D1']

注意事项

• source_id 与软件内部 DataSourceManager 一致（0=新浪 / 1=Tushare / 2=本地 / 3=腾讯）；

盘后下载专用的腾讯日线源不走 Manager，source_id 为 None。

• available 表示当前配置下的可用性，不做网络探测（调用本函数永远即时返回）。

数据源接入

本大类面向数据源接入者：想在吉宽量化里新接一个行情数据源（通达信 / 腾讯 / 新浪 / Tushare / 自建接口……）的用户。

将来所有与"数据源下载"有关的代码标准、接口要求、接口名称、输入输出变量都归入本大类。

为什么要按规范接入

盘后下载模块（下载窗口 / download_daily）与数据源是完全解耦的：

• 下载调度器只调用数据源暴露的统一接口，不关心数据是怎么拿到的；
• 数据源内部自己决定下载方式：支持批量就并发批量下，只支持单只就逐只下（网页接口自行加防反爬间隔）；
• 接口不符合规范的数据源，盘后下载不会识别、无法使用。

本大类文档

快速结论（详见规范文档）

一个数据源要被盘后下载认到，最少要有：

类属性 name / source_id / enabled / adjust_type 按实际值声明；输出 DataFrame 必须对齐 ../股票函数/格式标准.md。

相关文档

• ../股票函数/盘后下载与数据管理.md —— 盘后下载的使用方法（下载窗口 / download_daily / 覆盖检查）
• ../股票函数/格式标准.md —— 数据格式权威契约（所有数据源的输出必须完全对齐）

数据源接入 · 盘后下载接入规范

适用对象：想为自己的数据源接入吉宽量化盘后下载功能的接入者。

按本规范实现后，盘后下载窗口 / download_daily 会自动识别并使用你的数据源；

不符合本规范的数据源，盘后下载不认。

1. 架构原则：调度器与数据源完全解耦

盘后下载调度器（DailyDownloadCoordinator，位于 daily_downloader.py）只做四件事：

1. 拿品种列表 → 2. 调数据源的统一接口 → 3. 把返回的数据落盘 → 4. 汇报进度 / 处理取消。

下载方式由数据源自己决定，调度器完全不关心：

因此：下载策略（并发、限速、重试、翻页）全部写在数据源自己的文件里，调度器一行代码不需要改。

2. 文件与类的放置要求

3. 接口清单（核心）

3.1 必须实现的接口（缺一不可）

__init__(self) —— 构造函数

def __init__(self, enabled=True):
    super().__init__(name='你的数据源名', source_id=你的ID)

调用基类构造，传入显示名和唯一 ID。

is_available(self) -> bool —— 可用性检查【基类抽象方法，必须实现】

def is_available(self) -> bool:
    """依赖库未安装 / token 未配置时返回 False，盘后下载与数据源管理器会跳过它"""

get_data(self, code, period, ...) —— 按 count 取数【基类抽象方法，必须实现】

实时取数接口（图表 / 选股用）。盘后下载不直接调用它，但作为数据源必须提供。

get_data_by_range(self, code, period='D1', start_date=None, end_date=None, market_type='STOCK') -> DataFrame 或 None —— 按区间取一只【盘后下载的最小要求，必须实现】

def get_data_by_range(self, code, period='D1', start_date=None,
                      end_date=None, market_type='STOCK'):
    """
    取一只品种 [start_date, end_date] 区间内的历史数据。

    Parameters
    ----------
    code        : str   统一代码格式，带市场后缀：'000001.SZ' / '600000.SH' / '920000.BJ'
    period      : str   'D1' 日线 / 'W1' 周线 / 'MN' 月线 / '1min' '5min' '15min' '30min' '60min'
    start_date  : str   'YYYYMMDD'；None = 不限（从最早拉到最新）
    end_date    : str   'YYYYMMDD'；None = 不限
    market_type : str   'STOCK' / 'FUTURE' / 'INDEX'

    Returns
    -------
    pandas DataFrame  标准格式（见第 5 节），按时间升序
    None              该品种下载失败或无数据（不要抛异常！内部自己捕获）
    """

3.2 可选覆写的接口（批量下载能力）

iter_download_by_range(self, codes, period='D1', start_date=None, end_date=None, market_type='STOCK') —— 批量下载【可选】

不实现也能用——基类默认实现 = 逐只循环调用 get_data_by_range。

如果你的数据源有批量/并发能力，覆写本接口获得成倍提速：

def iter_download_by_range(self, codes, period='D1', start_date=None,
                           end_date=None, market_type='STOCK'):
    """批量下载：每完成一只就 yield 一条（生成器）"""
    # 1. 你的批量策略：分块、并发、限速……全部在这里实现
    # 2. 每完成一只：
    yield code, df        # df = 标准格式 DataFrame
    # 失败/无数据的品种也要 yield，df 传 None：
    yield code, None

约定：

• 生成器每完成一只就 yield (code, df) 一条，调度器边消费边落盘并更新进度；
• 每个传入的 code 都必须恰好 yield 一次（成功给 df，失败给 None），否则进度统计会错乱；
• 调度器中途取消时会调用生成器的 close()，你的内部循环会随之终止（批量在途的请求作废属正常）。

参考实现：DataSources/pytdxdata_source.py（分块 + 12 路并发 + 整块失败自动回退单只）。

4. 调度器实际调用流程（理解后自查）

品种列表 codes（'000001.SZ' 等统一格式）
   │  ① 已完全覆盖所选区间的品种自动剔除（skip_existing）
   ▼
datasource.iter_download_by_range(pending, period, start_date, end_date)
   │  ② 消费生成器，每条 (code, df)：
   │       df 是 None 或空  → 记入失败清单
   │       df 有效          → 自动补 ts_code 列后 saveSymbolData 落盘
   ▼
data/<market>/<period>/<symbol>.parquet + data_coverage_tb 自动更新

注意两点：

• 调度器调用 iter_download_by_range 时不传 market_type，各数据源用自己签名里的默认值即可；
• 落盘时读取你的类属性 adjust_type 标注复权口径——必须如实声明，否则增量合并会把不同口径的数据混在一起。

5. 输出格式标准（权威契约）

DataFrame 输出必须完全对齐 ../股票函数/格式标准.md，盘后下载相关列如下：

日线 / 周线 / 月线（period = 'D1' / 'W1' / 'MN'）

分钟线（period = '1min' / '5min' / ...）

期货日线（可选扩展列）

硬性要求：

1. 按 trade_date（或 trade_time）升序排列，已去重；
2. 失败/无数据返回 None，不要抛异常；
3. 复权口径由类属性声明，数据本身按声明的口径返回。

类属性 adjust_type 取值

6. 代码标准

1. 下载策略内聚：并发数、分块大小、请求间隔、防反爬抖动等全部作为数据源类内常量/属性，写在数据源文件里；
2. 失败不外抛：单只失败在数据源内部捕获并返回 None（批量生成器 yield None），不中断整个下载任务；
3. 每次会话新建连接对象（如有长连接库）：库的锁/事件对象绑定创建时的事件循环时，严禁跨会话缓存客户端实例；
4. 统一代码格式：对外接口一律使用带市场后缀的统一代码（'000001.SZ'），市场映射在数据源内部完成；
5. 日志走 print 之外请用项目 TPrint 体系（生产环境写入日志文件与日志面板）；
6. 每次请求的原始数据 → 标准格式的转换函数独立成方法（如 _to_df），便于测试与复用。

7. 接入步骤清单（照做即可）

1. 在 DataSources/ 新建 xxx_source.py，类继承 DataSourceBase；
2. 实现 __init__ / is_available / get_data / get_data_by_range；
3. （推荐）数据源有批量能力时，覆写 iter_download_by_range 生成器；
4. 如实声明类属性 name / source_id / enabled / adjust_type；
5. 在 DataSourceManager.py 的 _register_default_sources() 中注册实例；
6. 用盘后下载窗口选小范围（如"指定区间 1 个月"）试跑，检查：

• 进度正常推进、无卡死；
• 落盘文件列结构与 adjust_type 标注正确（用 check_data_coverage 验证）；
• 取消按钮能正常中断。

8. 完整示例模板（单只下载数据源）

# -*- coding: utf-8 -*-
"""
xxx 数据源适配器（示例模板）
"""
import pandas as pd
from .base import DataSourceBase


class XxxDataSource(DataSourceBase):
    """xxx 日线数据源（单只下载，逐只请求 + 防反爬间隔）"""

    source_id = 88                 # 唯一 ID，不与现有源冲突
    source_name = 'xxx 数据源'
    min_interval = 1.0             # 两次请求最小间隔（秒），网页接口必备
    adjust_type = 0                # 如实声明：0=未复权 1=前复权 2=后复权

    def __init__(self, enabled=True):
        super().__init__(name=self.source_name, source_id=self.source_id)

    def is_available(self) -> bool:
        return True                # 依赖缺失时返回 False

    # ---- 实时取数（基类抽象方法）----
    def get_data(self, code, period='D1', market_type='STOCK', count=1000,
                 save_to_local=False, overwrite=False):
        df = self.get_data_by_range(code, period=period, market_type=market_type)
        if df is None:
            return None
        return df.tail(count)      # 示例：取最近 count 根

    # ---- 盘后下载必须：按区间取一只 ----
    def get_data_by_range(self, code, period='D1', start_date=None,
                          end_date=None, market_type='STOCK'):
        try:
            raw = self._fetch_one(code)          # 你的 HTTP/TCP 请求，翻页拿全历史
        except Exception as e:
            print(f'[{self.source_name}] {code} 获取失败: {e}')
            return None
        if not raw:
            return None
        return self._to_df(raw, start_date, end_date)

    # ---- 原始数据 → 标准格式（独立方法，便于复用）----
    def _to_df(self, raw, start_date=None, end_date=None):
        rows = [{'trade_date': str(r['date']),
                 'open': float(r['open']), 'high': float(r['high']),
                 'low': float(r['low']),   'close': float(r['close']),
                 'vol': float(r['vol']),   'amount': float(r.get('amount', 0) or 0)}
                for r in raw]
        df = pd.DataFrame(rows)
        if start_date:
            df = df[df['trade_date'] >= str(start_date)]
        if end_date:
            df = df[df['trade_date'] <= str(end_date)]
        df.drop_duplicates(subset='trade_date', keep='last', inplace=True)
        df.sort_values('trade_date', inplace=True)
        df.reset_index(drop=True, inplace=True)
        return df if not df.empty else None

    # ---- 防反爬：单只数据源在请求层自行控制节奏 ----
    # （参考 DataSources/tencent_daily_source.py 的 _wait_interval 实现）
    def _fetch_one(self, code):
        ...

批量数据源示例（分块并发 + 生成器）：DataSources/pytdxdata_source.py 的

iter_download_by_range / _async_fetch_batch；

期货批量示例：DataSources/pytdxdata_future_source.py。

日期和时间

本文档规定软件内部统一使用的日期时间标准格式。

凡是不规范的日期、时间（无论来自数据源、用户输入还是外部文件），在软件代码中都必须先转换成标准格式再使用。

数据格式问题遵循数据提供端统一处理原则：在数据入口处转换，上层策略代码直接拿到标准格式。

全部转换逻辑收敛在统一中枢模块 datetime_utils 中——它是全软件唯一的日期时间转换出口，新代码禁止散落手写解析。

一、标准格式总表（权威规定）

要点：

• 日线和分钟线的区别只在有没有时间分量：日线 8 位纯数字，分钟线 12 位纯数字。分钟线取 bar 开盘时间（系统统一口径，与外盘/MT4 惯例一致）：'0930' 标签的 M5 bar 覆盖 09:30:00–09:35:00；国内数据源的结束时间标签在数据入口已统一转换（见 格式标准 第 0 节）。
• 同一份分钟数据在数据层是紧凑格式 YYYYMMDDHHMM，进入图表体系后带空格冒号 YYYYMMDD HH:MM:SS（由数据接口出口统一转换，见 格式标准）。两种写法不要混用。
• datetime.datetime 对象是内部计算的"中间语言"：任何格式先进来，算完再按场景输出。

二、转换函数

datetime_utils — 统一时间转换中枢（全软件唯一，推荐入口）

概述

所有日期时间转换逻辑的唯一所在地。策略代码可以直接 import（沙箱允许 import 软件模块），图表对象方法 isTimeStr 等内部也是委托它实现的。四个函数覆盖"解析 → 规范化 → 输出"全链路：

from datetime_utils import parse_any, normalize_time, fmt, parse_series

parse_any — 宽容解析

parse_any('20260924')              # → datetime.datetime(2026, 9, 24, 0, 0)
parse_any('2026-09-24')            # → datetime.datetime(2026, 9, 24, 0, 0)
parse_any('2026/09/24 09:31:00')   # → datetime.datetime(2026, 9, 24, 9, 31)
parse_any('abc')                   # → None（不抛异常，调用方判空）

normalize_time — 时间分量补齐

normalize_time('0930')      # → '09:30:00'
normalize_time('093000')    # → '09:30:00'
normalize_time('09')        # → '09:00:00'
normalize_time('09:30')     # → '09:30:00'（5位带冒号，补秒）
normalize_time('09:30:00')  # → '09:30:00'（已规范，原样返回）
normalize_time('9:30')      # → '9:30'（4位带冒号无法与已规范格式可靠区分，原样返回）
normalize_time('930')       # → '930'（3位无法推断，原样返回）

fmt — 按场景输出标准字符串

fmt(dt)                # → '20260924'           日线/交易日历（style='compact' 默认）
fmt(dt, 'minute')      # → '202609240930'       分钟线数据层
fmt(dt, 'chart')       # → '20260924 09:30:00'  图表分钟时间/画线对象
fmt(dt, 'clock')       # → '2026-09-24 09:30:00' 报价时钟拼接串

parse_series — 整列定长解析

df = self.MF.DC.get_market_data('600000.SH', 'D1', 100)   # 标准数据，trade_date 列
df['dt'] = parse_series(df['trade_date'], 'D1')           # 日线 '%Y%m%d'
df['dt'] = parse_series(df['trade_date'], '1min')         # 分钟 '%Y%m%d%H%M'

isTimeStr(timeStr) — 图表对象方法（等价入口）

概述

图表对象方法，内部调用 datetime_utils.parse_any。在画线对象编程、拿不到 import 的场景下用它；规则与 parse_any 完全一致：三种分隔符写法（无分隔符 / - / /）都认识，日期后可以不带或带 HH:MM:SS。

函数签名

self.MF.CP.isTimeStr(timeStr)

参数

返回值

• 合法 → datetime.datetime 对象
• 不合法（格式错、不是字符串）→ None（并写日志），调用方必须判空

示例代码

t1 = self.MF.CP.isTimeStr('20260924')              # 日线写法
t2 = self.MF.CP.isTimeStr('2026-09-24')            # 带横杠
t3 = self.MF.CP.isTimeStr('2026/09/24 09:31:00')   # 带斜杠和时间
if t1 is None:
    return(False)                                   # 非法输入，判空保护

输出结果

t1 = datetime.datetime(2026, 9, 24, 0, 0)
t2 = datetime.datetime(2026, 9, 24, 0, 0)
t3 = datetime.datetime(2026, 9, 24, 9, 31)

bar_end_to_open — 分钟 bar 结束标签 → 开盘标签（K 线标签统一口径）

概述

国内数据源的分钟 bar 以结束时间为标签（通达信约定），系统标准口径是 bar 开盘时间（与外盘/MT4 惯例一致，见 格式标准 第 0 节）。本函数把结束标签转成开盘标签，转换规则：结束时刻减去一个 bar 时长。

一般场景无需手工调用：落盘走 saveSymbolData（自动转换）、取数走 get_market_data 等出口（自动转换）；只有自接新数据源或处理原始数据时才需要显式调用。

函数签名

bar_end_to_open(time_str, period)      # 单值
bar_end_to_open_series(s, period)      # Series 整列
bar_minutes(period)                    # 周期时长（分钟数）

示例代码

from datetime_utils import bar_end_to_open, bar_end_to_open_series, bar_minutes

bar_end_to_open('202609240935', 'M5')          # → '202609240930'
bar_end_to_open('202609241500', 'M1')          # → '202609241459'
bar_end_to_open('202609241130', 'H1')          # → '202609241030'（A 股 H1 从 09:30 对齐的均匀 60 分钟）
bar_end_to_open('202609240000', 'H1')          # → '202609232300'（连续市场跨日自然处理）
bar_end_to_open('20260924', 'D1')              # → '20260924'（日/周/月线原样）
bar_minutes('M5')                              # → 5
s = bar_end_to_open_series(df['trade_date'], 'M5')   # 整列转换（带秒输入归一为 12 位）

pd.to_datetime — 直接定长解析（保留知识）

概述

中枢的 parse_series 已覆盖数据层整列解析。只有两种非紧凑消费格式还需要直接手写 pd.to_datetime（必须显式指定 format）：

示例代码

# 读报价时钟（date='%Y-%m-%d'，time='%H:%M:%S'），拼串转标准 datetime
_tickstr = ('%s %s' % (self.MF.CP.ticks['date'].get(),
                       self.MF.CP.ticks['time'].get())).strip()
self.ticktime = pd.to_datetime(_tickstr, format='%Y-%m-%d %H:%M:%S')

fmt(dt, style) / timedelta — 标准输出与日期偏移

概述

内部计算完成后，用中枢的 fmt 把 datetime.datetime 按场景输出成标准字符串（等价于手写 strftime，但格式名由中枢统一约定）；用 timedelta（月周期用 relativedelta）做日期偏移。

示例代码

import datetime
from datetime_utils import fmt
dt = datetime.datetime(2026, 9, 24, 9, 30)

fmt(dt)                # → '20260924'           日线标准
fmt(dt, 'minute')      # → '202609240930'       分钟标准
fmt(dt, 'chart')       # → '20260924 09:30:00'  图表标准
fmt(dt, 'clock')       # → '2026-09-24 09:30:00' 时钟标准

dt2 = dt + datetime.timedelta(days=1)               # 日期偏移 1 天

get_stock_calendar(exchange, start_date, end_date) — 交易日判断 / 日期偏移

概述

给定起止日期，返回其中所有交易日（自动剔除周末、节假日、休市日）。日历从本地上证指数 K 线推导，与回测、盘后下载共用同一套真实交易日。

函数签名

get_stock_calendar(exchange='SSE', start_date=None, end_date=None)

参数

返回值

list[str] — 升序 YYYYMMDD 交易日期字符串列表；本地无指数数据返回空列表。

示例代码

days = get_stock_calendar(start_date='20260901', end_date='20260930')

# 判断某天是否交易日
is_trade_day = '20260924' in days

# 取某天之后的下一个交易日
after = [d for d in days if d > '20260924']
next_trade_day = after[0] if after else None

# 统计区间交易日数量
print(len(days))

is_opentime(mytime) — 判断某时刻是否盘中开盘时间

概述

数据中心方法。判断一个 datetime.datetime 是否落在开盘时段内（先剔除非交易星期，再比对报价时段），程序化监控里判断"现在该不该干活"用这个。

函数签名

self.MF.DC.is_opentime(mytime)

参数 / 返回值

barShiftToTime(shift) / barTimeToShift(timeStr) — 图表K线位置与时间互转

概述

画线对象编程时使用：K线偏移值 ↔ 时间点双向转换。barTimeToShift 内部先用 isTimeStr 把字符串规范化，所以传入 20260924 / 2026-09-24 等写法都可以。

函数签名

self.MF.CP.barShiftToTime(shift)       # 偏移值 → 时间
self.MF.CP.barTimeToShift(timeStr)     # 时间字符串 → 偏移值

参数 / 返回值

三、常见不规范格式转换速查

时间分量补齐直接用中枢函数，不要自己再写：

from datetime_utils import normalize_time
normalize_time('0930')      # → '09:30:00'

四、完整示例：监控策略中的标准用法

import datetime
import pandas as pd
from datetime_utils import fmt

def OnTime(self, code):
    # 1. 读报价时钟（date='%Y-%m-%d'，time='%H:%M:%S'），拼串转标准 datetime
    _tickstr = ('%s %s' % (self.MF.CP.ticks['date'].get(),
                           self.MF.CP.ticks['time'].get())).strip()
    try:
        self.ticktime = pd.to_datetime(_tickstr, format='%Y-%m-%d %H:%M:%S')
    except ValueError:
        return(False)                # 报价未就绪（空字符串），本次跳过

    # 2. 判断是否盘中
    if not self.MF.DC.is_opentime(self.ticktime):
        return(False)

    # 3. 读标准数据：日线 trade_date='YYYYMMDD'，分钟='YYYYMMDDHHMM'
    df = self.MF.DC.get_market_data(code, 'D1', 100)
    if df is None or df.empty:
        return(False)

    # 4. 交易日过滤：非交易日直接跳过
    today = fmt(self.ticktime)                                    # → 'YYYYMMDD'
    if today not in get_stock_calendar(end_date=today):
        return(False)

    # ... 计算信号 ...
    return(False)

五、时区转换（国际盘标准）

标准规定

• 软件内部所有时间一律是北京时间（UTC+8），且为不带时区标记的 datetime.datetime（naive 墙钟时间）。这是全软件唯一的时间标准。
• 国际盘数据（黄金、外汇、外盘期货等）必须在数据入口处转换成北京时间再入库——延续"数据提供端统一处理"原则。策略、图表、下载器全程只见北京时间，无需各自处理时区。
• 需要展示国际市场当地开收盘时间时，在展示出口用 from_beijing 反向转换。
• 夏令时由 zoneinfo 自动处理（纽约冬 -5 / 夏 -4，伦敦冬 0 / 夏 +1），无需手工换算。

时区参数写法（三选一）

datetime_utils 时区函数一览

from datetime_utils import (tz_convert, to_beijing, from_beijing,
                            local_to_beijing, beijing_to_local,
                            tz_offset_hours, tz_delta_hours, local_tz_name)

示例代码

import datetime
from datetime_utils import to_beijing, from_beijing, tz_delta_hours

# 1. 国际盘数据入口：COMEX 黄金 2026-09-24 09:31（纽约当地）→ 北京时间入库
ny_time = datetime.datetime(2026, 9, 24, 9, 31)
to_beijing(ny_time, '纽约')          # → datetime(2026, 9, 24, 21, 31)（或 22:31，视夏令时）

# 2. 外汇数据源给的是 UTC：2026-09-24 12:00 UTC → 北京时间
to_beijing(datetime.datetime(2026, 9, 24, 12, 0), 'UTC')   # → datetime(2026, 9, 24, 20, 0)

# 3. MT4 服务器时间（GMT+3 固定偏移）→ 北京时间
to_beijing(ny_time, 3)               # → datetime(2026, 9, 24, 14, 31)

# 4. 展示出口：北京时间 21:31 → 纽约当地开收盘时间
from_beijing(datetime.datetime(2026, 9, 24, 21, 31), '纽约')

# 5. 时差查询：北京比纽约快几小时（冬令时 13，夏令时 12）
tz_delta_hours('北京', '纽约')

# 6. 任意两市场互转：伦敦 08:00 → 东京墙钟
from datetime_utils import tz_convert
tz_convert(datetime.datetime(2026, 9, 24, 8, 0), '伦敦', '东京')

注意事项

• 转换输入输出都是 naive 墙钟时间（不带 tzinfo），与软件既有约定一致，转换结果可直接入库、直接画图，不需要再 astimezone。
• to_beijing 只在数据入口调用一次；不要在策略里反复转换，否则会二次偏移。
• 拿不准数据源时区时，先用一根已知 K 线核时（如以国内夜盘开盘 21:00 为锚核对黄金数据），确认后再定 from_tz。

六、注意事项

• 全软件的日期时间转换只走 datetime_utils 中枢。新代码禁止散落手写 strptime / pd.to_datetime 转换逻辑；存量代码中已带显式标准 format 的调用保持不变，改动时顺势收编。
• 必须显式指定 format。pd.to_datetime(s) 不带 format 会按美式月/日猜测，'20260907' 这类纯数字还可能解析失败或出错，禁止使用。
• parse_any / isTimeStr 解析失败返回 None 而不是抛异常，调用后必须判空。
• 分钟 bar 时间标签是 bar 开盘时间（系统统一口径）：'202609240930' 表示 09:30:00 起的一根 bar（M1 覆盖 09:30:00–09:31:00，M5 覆盖 09:30:00–09:35:00）。国内数据源的结束时间标签在数据入口统一转换，转换规则与函数见 格式标准 第 0 节。
• 报价时钟未就绪时 ticks['date'] / ticks['time'] 为空字符串，直接拼串解析会抛 ValueError，要用 try / except 跳过本次。
• 数据层紧凑格式（YYYYMMDDHHMM）与图表带分隔格式（YYYYMMDD HH:MM:SS）是同一时刻的两种标准写法，按消费场景选对，不要互相混比。
• get_stock_calendar 依赖本地指数 K 线数据，本地无数据时返回空列表；先确认已下载上证指数日线。
• 月份偏移必须用 relativedelta(months=n)，timedelta(days=30) 不是一个月。
• 接入国际盘品种时，在数据源适配器入口统一调用 to_beijing，适配器出口之后全链路保持北京时间；时区参数写法见第五节。

技术指标

本文说明两件事：如何调用软件内置的标准指标（布林、CCI、KDJ、MA、MACD、OBV、RSI、SAR、ZIG），以及如何自己编写一个指标文件。

指标文件统一放在 公式库\指标类\ 文件夹，图表的指标加载功能会自动读取该文件夹下的 .py 文件。

一、调用标准指标

九个标准指标总表

在图表上调用

打开 K 线图表 → 加载指标 → 选择 公式库\指标类\ 下的指标文件 → 指标随 K 线绘制。主图指标叠加在 K 线上，副图指标显示在独立窗口。修改参数在指标设置中完成；参数若定义了下拉选项（见 Enum），设置界面以下拉菜单呈现。

在策略 / 程序中调用指标数据

有两种方法：简便调用（直接用内置指标函数，推荐）和完整调用（实例化指标类）。

方法一：简便调用内置指标函数（推荐）

软件在 Indicators 模块中内置了指标计算函数，一行 import、一行计算，直接拿到指标数据：

from Indicators import iMA, iMACD, iKDJ, MA, NMA, iSlope

df = self.MF.DC.get_market_data('600000.SH', 'D1', 100)   # 标准 K 线数据

ma = iMA(df['close'], 20, 0, 'SMA')      # 20 日均线（价格序列进，指标序列出）
dif = iMACD(df, 12, 26, 9)['dif']        # MACD 的 dif 线（K线数据进，新增指标列出）
kdj = iKDJ(df, 9, 3, 3)['k']             # KDJ 的 K 值

内置指标函数一览：

注意：iMACD / iKDJ / MA 会直接在传入的 DataFrame 上加列。不想污染原数据时先 df.copy()。

方法二：完整调用指标类

from 公式库.指标类.MACD import INDICATOR

ind = INDICATOR(self.MF, '600000.SH', 'D1')
ind.put['fastperiod'] = 12        # 可选：改参数（默认值已在类里）
ind.OnCalculate(df)
dif, dea, macd = ind.indicator_buffer1, ind.indicator_buffer2, ind.indicator_buffer3

两种方法的区别

两种方法计算结果完全一致——指标类内部就是调用这些内置函数来算的。策略中只取值做信号判断，用方法一即可。

二、编写自定义指标

指标文件骨架

# -*- coding: utf-8 -*-
from Indicators import iMA

class INDICATOR:                          # 类名必须是 INDICATOR
    def __init__(self, MF, symbol, period):
        self.IndicatorName = 'MYIND'      # 指标名称（图上显示）
        self.indicator_chart_main = False # 主图还是副图
        self.indicator_plots = 1          # 输出线数量

        self.indicator1 = ['line', 'MY', 'white', 1]     # 第 1 条线的样式
        self.indicator_buffer1 = []                      # 第 1 条线的数据
        self.put = {}                                    # 输入参数
        self.put['N'] = 14
        self.Enum = {}                                   # 下拉菜单参数（可选）
        self.period = period                             # 保存传入的周期
        self.symbol = symbol                             # 保存传入的代码

    def OnCalculate(self, Data):
        data = Data                                      # K 线 DataFrame
        self.indicator_buffer1 = iMA(data['close'], self.put['N'], 0, 'SMA')

把文件保存到 公式库\指标类\（文件名即指标加载名），在图表中加载即可。

标准属性逐项说明（每一项软件都提供标准化格式）

标准函数库（from Indicators import ...）

软件自带的指标计算函数，指标文件里直接 import 使用，不要自己重复造轮子。函数的完整签名、输入输出见上文"方法一：简便调用内置指标函数"中的函数一览表——iMA / NMA / iSlope 传入价格序列、返回指标序列；iMACD / iKDJ / MA 传入 K 线 DataFrame、在其上新增指标列后返回。

完整示例：带下拉参数和水平线的副图指标

# -*- coding: utf-8 -*-
from Indicators import iMA

class INDICATOR:
    def __init__(self, MF, symbol, period):
        self.IndicatorName='MYOSC'
        self.indicator_chart_main=False                 # 副图
        self.indicator_plots=2
        self.indicator_level=[[0],['#800000'],[':'],[0.8]]   # 零轴水平线
        self.indicator1=['histogram','osc','red',0.2]        # 柱
        self.indicator2=['line','ma','white',1]              # 均线
        self.indicator_buffer1=[]
        self.indicator_buffer2=[]
        self.put={}
        self.put['N']=12
        self.put['ma_method']='EMA'
        self.Enum={'ma_method':['SMA','EMA','SMMA']}    # 下拉菜单
        self.period=period
        self.symbol=symbol

    def OnCalculate(self, Data):
        if not isinstance(self.put['N'], int) or self.put['N'] <= 0:
            print('参数N必须为大于0的整数')
            return
        data = Data
        osc = data['close'] - iMA(data['close'], self.put['N'], 0, self.put['ma_method'])
        self.indicator_buffer1 = osc.copy()
        self.indicator2[2] = ['red' if v > 0 else 'lime' for v in self.indicator_buffer1]  # 柱逐根变色
        self.indicator_buffer2 = iMA(osc, 5, 0, 'SMA')

注意事项

• 指标文件修改后重新加载指标即可生效（热更新），无需重启软件。
• 副图建议用 indicator_minmax 固定纵轴（如 0～100 类指标），否则随数据缩放不便观察。
• OnCalculate 内不要修改传入的 Data 的原始列；需要衍生列时先 data = Data 后新增列（如 CCI 的 typ），或 Data.copy()。
• 指标加载与执行同用户策略一样受沙箱保护：只能调用 / 读取软件模块，不能改写软件对象。

文件函数（读写本地文件）

文件函数是用户代码（策略 / 选股 / 指标）读写本地文件的统一接口。

所有文件操作都通过 import api 后调用 api.* 完成，路径越界、危险类型会在函数内部直接拦截并给出中文提示，而不是等沙箱报错。

允许写入的位置只有两类：用户数据区（软件目录内的专属文件夹）和软件目录之外的任意路径。

一、用户数据区与路径规则

用户数据区是软件目录下的 userdata\ 文件夹，专门给用户代码存放文件。它和用户配置、公式库一样属于升级 / 重装时保留的内容，放心把中间结果长期放在这里。

其它约定：

• 统一编码：文本 / JSON 用 UTF-8；CSV 用 UTF-8 带 BOM（utf-8-sig），Excel 双击打开中文不乱码。
• 路径分隔符 / 和 \ 都可以，Windows 与网络路径均支持。
• 写类函数的上级文件夹不存在时自动创建，无需先 make_dir。

二、函数一览

三、文本族 read_text / write_text

概述

保存 / 读取纯文本：策略运行日志、信号流水、自己生成的报告。write_text 加 append=True 即追加写入，适合按日累计的记录。

函数签名

api.read_text(file, encoding='utf-8')
api.write_text(file, text, append=False, encoding='utf-8')

参数

返回值

• read_text：文件全部内容（字符串）；文件不存在打印提示并返回 None
• write_text：成功 True；路径越界 / 危险类型直接报错

示例代码

import api

# 每次触发信号时追加一行记录
api.write_text('signals.log', '2026-10-04 09:31  600000.SH  金叉买入\n', append=True)

# 收盘后整体读回
log = api.read_text('signals.log')
if log is not None:
    print(log[:80])            # 只打印前 80 个字符

输出结果

2026-10-04 09:31  600000.SH  金叉买入

四、DataFrame 族 read_csv / write_csv

概述

量化场景最常用的一对：把回测结果、选股名单、指标数据存成 CSV，用 Excel 直接打开；或把外部整理好的数据表读回策略。底层即 pandas 读写，read_csv 额外参数（usecols、dtype、parse_dates…）原样透传。

函数签名

api.read_csv(file, **kwargs)
api.write_csv(file, df, index=False, encoding='utf-8-sig')

参数

返回值

• read_csv：DataFrame；文件不存在返回 None
• write_csv：成功 True；空表也允许写入（只含表头）

示例代码

import api

# 导出选股名单（股票代码用标准格式）
pool = api.get_stock_namelist(market='SH')
api.write_csv('my_pool.csv', pool)

# 下次运行直接读回，不必重新下载
df = api.read_csv('my_pool.csv')
if df is not None:
    print(len(df), '只股票')

输出结果

1200 只股票

五、配置族 read_json / write_json

概述

保存结构化的参数和状态：策略的自定义参数、跨重启需要恢复的持仓标记等。dict / list 与 JSON 文件一一对应，中文原样保存。

函数签名

api.read_json(file)
api.write_json(file, obj, indent=2)

参数

返回值

• read_json：文件对应的 dict / list；文件不存在返回 None；内容不是合法 JSON 直接报错
• write_json：成功 True

示例代码

import api

# 策略参数存档
cfg = {'fast': 5, 'slow': 20, 'pool': 'my_pool.csv'}
api.write_json('strat_cfg.json', cfg)

# 下次运行恢复（文件不存在则用默认值）
cfg = api.read_json('strat_cfg.json') or {'fast': 5, 'slow': 20, 'pool': ''}
print(cfg['fast'], cfg['slow'])

输出结果

5 20

六、文件与目录管理 file_exists / file_delete / list_files / make_dir

概述

对用户数据区内的文件做基础管理：判断存在性、按通配符列文件、清理过期的中间结果。

函数签名

api.file_exists(file)
api.file_delete(file)
api.list_files(folder='.', pattern='*', recursive=False)
api.make_dir(folder)

参数

返回值

• file_exists：True / False
• file_delete：成功 True；文件不存在 False；用户数据区之外传绝对路径直接拒绝
• list_files：完整路径字符串列表（已排序）
• make_dir：成功 True（文件夹已存在时同样返回 True）

示例代码

import api

# 清理上个月的临时 CSV，只留最新的
for f in api.list_files(pattern='tmp_*.csv'):
    if '202609' in f:
        api.file_delete(f)
        print('已删除', f)

print(api.file_exists('my_pool.csv'))

输出结果

已删除 D:\stock\userdata\tmp_pool_20260901.csv
True

七、允许与禁止（安全边界）

允许：

禁止：

八、注意事项

• 不要在 OnTick 里高频写盘：每秒写文件会明显拖慢策略，日志类写入建议放在信号触发时或收盘后。
• 程序化监控与回测避免同时写同一个文件：两边同时写会互相覆盖，需要共享的数据请分文件存放。
• file_delete 删除不可恢复，删除前可用 file_exists 确认、把重要文件先 read_text 读回备份。
• 文件不存在时读类函数返回 None 而不是报错，方便策略轮询外部数据是否就绪；路径越界、类型非法则直接报错，按提示改路径即可。
• 直接用 open() 写软件目录之外的文件沙箱并不拦截，但统一走文件函数可获得路径校验、编码统一、自动建目录这三层保障，推荐全部走接口。

时间序列和指标访问（K 线时序与运行机制）

讲清楚软件里 K 线时间序列怎么驱动策略运行：回测引擎逐 K 线推进、每次给策略看什么数据、data 窗口的边界、以及当前根 / 上一根的取值惯例。

指标怎么计算和调用见技术指标，本文只讲"时序"。

一、两种执行入口（软件设计约定）

策略通过两个互补的入口保证代码持续有执行机会：

TickSymbol：OnTick 的宿主品种声明

OnTick 源于"策略依附一个品种"的报价驱动模式，而本软件的策略是独立运行、可同时监控多品种的——多品种时 OnTick 无法确定依附谁。所以约定：

class _S(Strategy):                    # 监控/实盘策略
    TickSymbol = '600000.SH'           # 声明宿主品种（标准代码，一个策略只能写一个）
    Timer      = 60                    # 声明 OnTime 轮询间隔（秒，必填）
    def OnTick(self, code, data):      # code 恒为 TickSymbol；data 为该品种 D1 实时K线窗口
        ...
    def OnTime(self, code):            # 时钟驱动，照常轮询全部监控品种
        ...

• 声明了 TickSymbol：以该品种的实时报价驱动 OnTick（最高每秒一次），code 恒为该品种，data 是它的 D1 实时 K 线窗口；该品种必须包含在本程序监控范围内
• 未声明 TickSymbol：OnTick 在监控/实盘中不生效，只走 OnTime（选股/扫描模式，属正常用法）
• Timer 必须显式声明：未声明或不是有效正数 → OnTime 不生效
• 启动检测：监控程序启动时对声明做检测，不合规的入口不生效并在日志面板输出原因，例如：
• TickSymbol 写了多个品种（如 '600000.SH,000001.SZ'）→ OnTick 不生效，提示"只允许声明一个宿主品种"
• TickSymbol 不是标准代码字符串 / 不在监控范围 → OnTick 不生效，提示原因
• 未声明 Timer / Timer = 0 或负数 → OnTime 不生效，提示"未声明 Timer（轮询间隔秒数）"
• 两个入口都未生效 → 额外警告"没有任何生效的执行入口，监控将空转"
• 回测不受影响：回测一律走 OnTick（挂载品种逐 K 线回放），TickSymbol / Timer 不参与回测

适用边界：单品种实时量化（黄金、股指、单票网格）用 TickSymbol + OnTick——报价级即时反应；选股扫描、全市场监控（几十到几千只）不声明 TickSymbol，用 OnTime 定时批量扫描。两个入口可以同时实现：行情活跃时 OnTick 高频触发，行情沉寂时 OnTime 定时兜底。这套强制声明的目的，是让策略作者在运行前就主动明确自己需要哪种驱动模式、驱动的品种是什么、轮询间隔是多少。

二、OnTick 的 data：截至当前根的数据窗口

回测引擎逐根推进 K 线，每根调用一次 OnTick(code, data)。data 是从历史开头累积到当前这根 K 线的数据——包含当前根，绝不含未来数据（无未来函数是引擎保证的，不需要策略自己防）：

def OnTick(self, code, data):
    self.data = data                      # 策略内保存，供其他方法使用
    # data 行数随回测推进从少到多增长
    price = data.close.values[-1]         # 当前根收盘价（下单价惯例）
    prev  = data.close.values[-2]         # 上一根收盘价（金叉/死叉判断惯例）

推论：

• values[-1] 永远是当前正在处理的这根 K 线；values[-2] 是上一根
• 策略开头做数据长度检查：if data is None or len(data) < N + 2: return False（N 为最大指标周期，+2 保证 [-2] 不越界、均线有值）
• 图表上的 K 线、指标、买卖标记与 data 同步生长——引擎每推进一根，图表加一根，策略算一次

三、每根 K 线内部的处理顺序

引擎每根 K 线按固定顺序执行：

1. 把当前根追加进 data 窗口和图表
2. 自动止损止盈检查（autoCloseOrder）：用当根最高/最低价
   判断已设 stoploss / takeprofit 的持仓是否触发，触发即自动平仓
3. 调用 OnTick(code, data) → 策略发单（send / close）
4. 记录当根权益曲线

要点：止损止盈在策略代码之前检查——同一根 K 线里如果先触及止盈价、策略又发出新信号，先执行的是止盈平仓。所以设置 stoploss / takeprofit 后不要在策略里重复写平仓逻辑，两者会打架。

四、data 的标准列与时间列

data 是 pandas DataFrame，列名固定（格式标准）：

时间列的解析、偏移、交易日判断等操作一律走日期和时间的标准函数，不要手工切字符串。

五、K 线周期

data 的周期由策略挂载的图表周期决定；程序化读取用 get_market_data(code, period)（行情数据读取）。周期统一使用系统标准表述（权威契约见格式标准第 0 节）：

数据源原生周期写法（1min/day/MIN_5 等）只允许出现在各数据源的入口映射表值里，进系统即转成标准周期；'M1' 永远表示 1 分钟，月线是 'MN'。K 线时间标签统一为 bar 开盘时间（'0930' 标签的 M5 bar 覆盖 09:30:00–09:35:00，与外盘/MT4 惯例一致，详见格式标准第 0 节）——外盘/外汇/黄金数据天然同口径，无需再转换。

六、pandas 时序访问惯例（策略内推荐写法）

c = data.close                    # 收盘价 Series（带行号索引）
c.values[-1]                      # 当前根（最快，推荐）
c.iloc[-2]                        # 上一根
data.tail(5)                      # 最近 5 根完整行
data[data.volume > 0]             # 条件筛选
ma = iMA(data.close, 20, 0, 'EMA')   # 指标序列与 K 线一一对应（见技术指标文档）
ma.values[-1]                     # 当前根指标值

• 指标序列（iMA / iMACD 等）返回值与 data 行对行对齐：第 i 个指标值对应第 i 根 K 线，所以"当前根指标值"就是 指标.values[-1]
• 交叉判断固定写法：fast.values[-1] > slow.values[-1] and fast.values[-2] <= slow.values[-2]（当前根上穿 + 上一根未上穿）
• 需要"跳出当前窗口"按时间点或按根数取历史某根 K 线（时间 + 开高低收 + 成交量）——api.get_stock_kline 的两种标准取法见行情数据读取"取某一根 K 线的两种标准方法"

七、监控 / 实盘中的两个入口

实时监控与实盘交易中，引擎同时调度两个入口（回测中 OnTime 失效）：

报价驱动（OnTick）：策略类声明 TickSymbol = '600000.SH'（唯一宿主品种，须在监控范围内）时生效——实时行情内存表每收到该品种新报价，立即调用 OnTick(TickSymbol, data)，data 是它的 D1 实时 K 线窗口（含当日实时 bar，get_bars 输出与盘后下载同一套标准格式），最高每秒触发一次；未声明则 OnTick 不生效。异常熔断：连续 10 次抛异常暂停 OnTick 并打日志。

时钟驱动（OnTime）：不管行情是否活跃，每隔 Timer 秒对全部监控品种轮询调用一次 OnTime(code)。Timer 必须在策略类中显式声明（未声明或无效则 OnTime 不生效并打日志）。策略内部自己取最新数据判断（惯例取日 K）：

def OnTime(self, code):
    D = self.cp.MF.DC.get_market_data(code, 'D1')   # 最新全量日 K
    if D is None or len(D) < 22:
        return False
    fast = iMA(D.close, 5, 0, 'EMA')
    ...
    return code        # 返回 code = 出信号，交给监控回调播报

OnTick / OnTime 返回非 False / None 值（如返回 code）即视为出信号，交给监控回调播报。监控场景的信号去重、播报节流由监控机制负责，详见程序化监控策略开发。

八、图表对象的时间定位（简要）

画买卖箭头、画线时把 trade_date 时间换算成图表横轴位置：self.cp.barTimeToShift(时间)；反向换算 barShiftToShift / 取图表数据见图表绘图 API。回测下单时引擎自动用这些函数把买卖标记画在触发信号的那根 K 线上，策略无需处理。

注意事项

• 不要在策略里引用 data 之外的未来数据（如直接读全量文件、"预测"下一根），引擎保证的"无未来"只覆盖 data 窗口
• values[-1] 与图表显示的最后一根 K 线是同一根；回测进行中它随推进不断变化，不要在 OnTick 里缓存跨根使用的"当前价"
• 分钟周期 trade_date 是 bar 开盘时间：09:30 标签的分钟 bar 覆盖 09:30:00–09:35:00（M5，即从标签时刻开始的下一根 bar 之前）
• 周线/月线由日线合成，耗时操作（如长周期指标）建议直接在日线数据上算再自行聚合
• 监控 / 实盘中 OnTick 的 data 是 D1 实时窗口；策略要看分钟数据可自行从实时内存表取：self.cp.realtime_buffer.get_bars(code, 'M5')（周期同监控预热：M1/M5/M15/M30/H1/D1）
• OnTick 每秒快照驱动一次，策略不要在 OnTick 里写 sleep 或循环等待；长时间的处理会推迟下一次执行