SDK 使用手册
从样本证据到逐笔决策的完整闭环(v0.1.4,逐段可运行)
1 安装与分层
pip install edgekit # 核心:纯标准库,import 不拉起任何第三方依赖
pip install edgekit[tooling] # 想用离线模拟工具(x_sweep/calibrate_h/dd_check)才需要 numpy- core(solver / monitor / engine):纯确定性,禁随机、禁 IO,同输入同输出。
- tooling:离线模拟,seed 必填,同 seed 同结果;
x、h这类”必须标定”的参数由它产出。 - policy:把 tooling 的产物固化成一份冻结配置,再喂给 core。
整条闭环:solver 算执行参数 → tooling 标定 x/h → 组装成 Policy → engine 逐笔定仓 → monitor 逐笔检测失效 → INVALID 即停、人工复核后签发新 Policy 回到起点。
2 第 1 步:solver —— 该不该做、单笔冒多大险
solve 收的不是”胜率 0.6”,而是”100 笔里赢 60 笔”——没有样本量就没有凯利。它用置信下界(越不确定越保守)而非点估计。
import edgekit
r = edgekit.solve(wins=60, losses=40, RR=1.5)
r.edge_gate # True —— 置信下界仍有正优势,放行
r.risk_fraction # 0.02 —— 推荐的单笔风险比例(被 risk_cap 封顶)
# report 是一条可审计的证据链,不是点估计算出的虚假精确:
r.report.p_hat # 0.6 点估计胜率
r.report.p_lb # 0.5361 Wilson 单边置信下界
r.report.rr_lb # 1.1547 盈亏比对数正态下界
r.report.kelly_point # 0.1667 点估计半凯利
r.report.kelly_lb # 0.0672 下界半凯利(递减链:0.167 → 0.067 → 封顶 0.02)
r.report.binding_constraint # "risk_cap" 最后是谁说了算
r.report.cap_cost_ratio # 0.2762 封顶换走了多少增长 g(capped)/g(f_LB)
r.report.sensitivity # p/RR 各 ±5% 时未封顶弹性:p_up=0.0955 p_down=0.0390 ...证据不足会优雅降级,不抛错:wins=0 或 losses=0 → edge_gate=False,report 注明原因。结构非法(负数、RR<=0、参数越界)才在入口抛 ValueError/TypeError。
3 第 2 步:tooling —— 标定 x 和 h(离线,需 numpy)
x(上调门槛)和 h(CUSUM 报警线)没有默认值,必须由模拟标定后写进 policy。三个工具的 seed 必填,结果带 provenance(seed/dgp/tool_version)。
from edgekit.tooling import x_sweep, calibrate_h, dd_check
# x_sweep:在 x∈{1.1..3.0} × d∈{0.70..0.90} 上联合选,存活率≥95% 里取终值对数中位数最大者
xs = x_sweep(p0=0.6, RR0=1.5, cv_win=1.0, DD_hard=0.30,
seed=7, n_paths=2000, n_trades=500, risk_fraction=r.risk_fraction)
xs.x, xs.d # 2.9, 0.9(示例规模;生产用 10000×500)
xs.survival_rate # 1.000
xs.provenance.tool_version # "0.1.4"(跟随包版本)
# calibrate_h:找出让健康策略平均至少 ARL_0 笔才误报一次的最小 h
ch = calibrate_h(p0=0.6, RR0=1.5, cv_win=1.0,
seed=7, n_paths=2000, n_trades=500, ARL_0=200)
ch.h, ch.measured_arl # 11.0, 202.7(≥ 目标 200)
# dd_check:按 policy 实际参数校核 DD_hard 的误触发概率
dd = dd_check(p0=0.6, RR0=1.5, cv_win=1.0, DD_hard=0.30, x=xs.x, d=xs.d,
seed=7, n_paths=2000, n_trades=500, risk_fraction=r.risk_fraction)
dd.false_trigger_prob # 0.0000工具的输出随 seed 和样本量变化。n_paths/n_trades 这里取小值是为了示例跑得快;标定生产参数时用 PRD 规格的 10000×500(v0.1.2 起已向量化,秒级可跑)。
4 第 3 步:Policy —— 固化成冻结配置
把 solver 的结果、tooling 标定的 x/d/h、以及溯源信息打包成一份 Policy。它是 frozen dataclass,载入时校验取值区间,JSON 往返无损。
edge_0 = 0.6 * 1.5 - (1 - 0.6) # 基线优势 = 0.5
policy = edgekit.Policy(
p0=0.6, RR0=1.5, n0=100,
kelly_fraction=0.5, risk_cap=0.02, confidence=0.90,
risk_fraction=r.risk_fraction, # solver 产出
k=edge_0 / 2, h=ch.h, DD_hard=0.30, n_min=20, # monitor 参数,h 来自 calibrate_h
x=xs.x, d=xs.d, # rebase 参数,来自 x_sweep
portfolio_cap=0.06, gap_multiplier=1.0,
sim_seed=7, dgp=xs.provenance.dgp,
tool_version=xs.provenance.tool_version, policy_version="v1",
)
edgekit.save_policy(policy, "policy.json") # 唯一碰文件系统的薄壳
policy = edgekit.load_policy("policy.json") # 载入即校验;越界字段抛 ValueErrorparse_policy(json_str) / policy.to_json() 是不碰文件系统的纯函数版本,往返逐字段无损。
5 第 4 步:engine —— 把风险比例换算成仓位
风险的唯一含义:止损被打掉时账户亏的钱。position_size 由止损距离反推仓位,给三个口径。
# 现货(默认 leverage=1):止损 5% → 名义仓位 40000,止损打掉恰好亏 risk_budget
ps = edgekit.position_size(base=100000, risk_fraction=policy.risk_fraction,
risk_multiplier=1.0, stop_distance=0.05)
ps.risk_budget # 2000 = base × risk_fraction,止损时的最大亏损
ps.notional # 40000 敞口口径:notional × stop_distance == risk_budget 恒成立
ps.margin # 40000 保证金口径 = notional / leverage(leverage=1 时相等)
ps.contracts # 40000 合约张数 = notional / contract_multiplier
# 杠杆/期货:leverage 与 contract_multiplier 只换口径,不放大止损亏损
ps = edgekit.position_size(base=100000, risk_fraction=0.02, risk_multiplier=1.0,
stop_distance=0.05, leverage=3, contract_multiplier=10)
ps.notional # 40000 敞口不变(风险不变式不破)
ps.margin # 13333.3 = notional / 3,保证金随杠杆减少
ps.contracts # 4000 = notional / 10
# 账户级约束与再基准化
edgekit.portfolio_check(existing=[0.02, 0.02], new=0.02, cap=0.06) # allowed=True, remaining=0.02
edgekit.rebase(base=100000, equity=80000, x=policy.x, d=policy.d) # 跌破 d·base → base 下调到 80000leverage < 1、contract_multiplier <= 0、stop_distance <= 0、gap_multiplier < 1 都抛 ValueError,不裸崩。
6 第 5 步:monitor —— 逐笔检测失效,把降档接回仓位
每笔平仓后把标准化盈亏 X_t 喂给 monitor_step,它跑两道独立防线(回撤硬熔断 + 单边 CUSUM),返回 state 与 risk_multiplier;后者乘进下一笔的 position_size。统计量从 S_prev 进、S_t 出,monitor 自己不存状态。
from edgekit.monitor import Baseline, Snapshot
bl = Baseline(p0=0.6, RR0=1.5, n0=100, h=5.0) # h 示意;生产用 calibrate_h 产出的值
base, S, n = 100000, 0.0, 25 # 样本已过 n_min=20
for label, X_t, dd in [("健康", 1.5, 0.05), ("连亏", -1.0, 0.08), ("连亏", -1.0, 0.12),
("连亏", -1.0, 0.18), ("连亏", -1.0, 0.24)]:
m = edgekit.monitor_step(bl, Snapshot(X_t=X_t, DD_current=dd, n=n), S)
S, n = m.S_t, n + 1
ps = edgekit.position_size(base=base, risk_fraction=0.02,
risk_multiplier=m.risk_multiplier, stop_distance=0.05)
# 健康: S_t=0.00 OK risk_mult=1.0 notional=40000
# 连亏: S_t=1.25 OK risk_mult=1.0 notional=40000
# 连亏: S_t=2.50 DEGRADED risk_mult=0.5 notional=20000 ← 越过 h/2,风险减半
# 连亏: S_t=3.75 DEGRADED risk_mult=0.5 notional=20000
# 连亏: S_t=5.00 INVALID risk_mult=0.0 notional=0 ← 越过 h,停手
...四种状态:OK / DEGRADED(减半)/ INVALID(停手)/ INSUFFICIENT_DATA(n < n_min,不可判定,既不宣告正常也不宣告失效)。回撤硬熔断不看样本量、第 1 笔就生效:
m = edgekit.monitor_step(Baseline(p0=0.6, RR0=1.5, n0=100, h=5.0),
Snapshot(X_t=-1.0, DD_current=0.31, n=1), 0.0)
m.state, m.risk_multiplier # "INVALID", 0.0 —— DD_current ≥ DD_hard,硬线最优先传入 p_hat、RR_hat 可顺带拿到归因诊断(胜率单边下侧检验、盈亏比 ln RR 置信上界),只供人工复核,不参与判定。
PAUSE 之后没有自动恢复:人工看过归因、用新基线签发新版 policy(新的 h 重新标定、统计量归零),回到第 1 步——闭环。
可选:用 MonitorSession 托管净值高点与 S_t。 上面的循环要你手维护两样东西:把 S 从上一笔穿到下一笔,以及自己算 DD_current。后者最容易出错——DD_current = 1 − 当前净值 / 净值历史高点,分母是净值历史高点(running 高点),不是 base。base 慢上快下、通常小于高点,拿它当分母会系统性低估回撤,DD_hard 熔断可能永不触发,而库察觉不到。MonitorSession 把这两样收过来:构造时给初始净值(即首个高点),之后每笔只喂 equity / X_t / n,它内部更新 running 高点、按上式算 DD_current、穿线 S_t,判定仍全转交那个一字未改的纯 monitor(),返回同样的 MonitorResult。
from edgekit.monitor import Baseline
bl = Baseline(p0=0.6, RR0=1.5, n0=100, h=5.0)
sess = edgekit.MonitorSession(bl, initial_equity=100000) # 起始净值即首个高点
# 之后每笔只喂 equity / X_t / n;高点、DD_current、S_t 全由 session 维护
sess.update(equity=101000, X_t=1.5, n=25) # 创新高 → peak=101000,DD_current=0
sess.update(equity=99000, X_t=-1.0, n=26) # DD_current = 1 − 99000/101000 ≈ 0.02
m = sess.update(equity=95000, X_t=-1.0, n=27)
(m.state, m.risk_multiplier, round(m.S_t, 2))
# ("DEGRADED", 0.5, 2.5) —— 与手工 monitor_step(同 X_t 序列、自算 dd、自穿 S)逐笔一致X_t 口径不变:仍是按满额风险(base × risk_fraction,risk_multiplier=1 时)归一的 R 倍数,哪怕当前已 DEGRADED、实际只下半仓,也不按减半后的风险额归一——否则与喂给 calibrate_h 的序列口径不一,标定出的 h 失真。MonitorSession.update 的 docstring 把这条钉死了。
7 小结
solve(样本) → x_sweep/calibrate_h(标定 x,h) → Policy(冻结) → position_size(定仓) → monitor_step(检测)
↑__________ INVALID → 人工复核 → 新 policy __________|
整个库就是这条闭环:先用样本证据算稳健参数,离线标定不可省的两个参数,固化成策略,逐笔定仓,持续监控,失效即停、重审再启。它不寻找优势、不自动交易、不在运行中调参——你的全部自由裁量在”找优势、定止损”,之后的执行参数交给数字。