Advanced 研究工作簿

研究 ID / 作者 / 日期 / 版本：
主问题与机制：谁付钱，承担什么风险？
Observation / Mechanism / Hypothesis：
原假设 H0 / 反驳条件 / 经济意义下限：
研究单位 / 持有期 / 重叠处理：
数据来源 / 获取时间 / 首次可见时间 / 版本 / 许可：
资产池 / 退市 / 缺失 / 标签修订：
训练、验证、最终测试的起止点：
全部尝试与失败记录 / 参数选择规则：
基线 / 消融 / 成本压力：
净收益分布 / 样本数 / 区间 / 尾部：
时序依赖 / 多重试验 / 未覆盖机制：
结论：支持、推翻或证据不足；引用具体输出：
下一步：继续研究、Testing、暂停或退役；负责人：
输入文件 / 代码版本 / 运行命令 / 输出校验：

Strategy ID: 稳定且唯一的标识
Name: 可读名称与版本
Market: 市场、场所和时区
Instrument: 标的、合约规格、币种与每条腿
Horizon: 信号频率、持有上限、标签区间
Category: 收益机制分类；另注明应用场景
Hypothesis: 可被否定的主张及结果指标
Mechanism: 参与者、约束、为何有补偿
Data: 来源、可见时间、版本、缺失与修订处理
Features: 仅用历史的计算窗口、变换和单位
Signal: 明确逻辑、阈值、无效输入时的行为
Entry: 何时允许提交、可执行价格与未成交处理
Exit: 正常退出、时间退出、失效退出和未知订单
Sizing: 目标名义、风险缩放与硬上限
Risk: 单策略、组合、场所、融资和停机限制
Execution: 订单方式、子单、时限、延迟与余量
Cost: Fee、Slippage、融资等分项，禁止重复
Backtest: 输入、代码、基线、成本、完整输出
OOS: 隔离测试边界、泄漏检查和失败样本
Regime: 当时可知状态、适用边界、默认行为
Failure: 哪些证据推翻机制，何时暂停
Capacity: 正常与压力容量、资本和退出约束
Status: 生命周期状态、批准者、日期和证据

对象 ID / 版本 / 对应 Memo：
数据清单与校验值 / 时钟与可见性：
资产池历史 / 缺失和退市处理：
规则、参数、种子、全部试验数量：
成交与资金假设 / 初始资本 / 费用：
训练、验证、测试边界 / Purge / Embargo 依据：
毛收益、净收益、最大回撤、换手与暴露：
现金或持有基线 / 单信号消融：
参数邻域 / 状态 / 成本 / 容量 / 跳空压力：
逐期权益 / 成交 / 未成交 / 拒绝 / 现金流水：
统计区间与依赖假设 / 不支持的风险：
重算步骤 / 预期结果 / 失败时停止条件：
结论与晋级决定，含独立评阅意见：

报告时间 / 数据截止 / 策略版本 / 估值币种：
总资本 / 已分配 / 现金 / 可用保证金：
逐策略净敞口、总敞口、风险预算、场所：
因子定义 / Beta / 其他因子 / 非线性暴露：
协方差窗口 / 相关矩阵 / 风险贡献和加总：
波动目标 / 实际波动 / 杠杆与换手上限：
压力情景输入、损失、资金缺口与可退出量：
回撤 / 已触发档位 / 订单与持仓剩余风险：
集中度 / 对手方 / 数据与执行共同依赖：
缺口 / 责任人 / 立即行动 / 下次复核：
批准或暂停决定 / 恢复条件 / 独立批准者：

母单 ID / 方向 / 数量 / 决策时间 / 决策价：
到达时间与基准 / 最晚结束时间 / 终场标价：
三种执行规则与共同输入，事前冻结：
每笔子单：时间、场所、数量、成交、余量、费用：
总成交 / 余量 / 均价 / 到达价偏差 / 决策价偏差：
机会成本 / 完成时间 / 逆向选择 / 在途风险：
价格路径、深度、费用、延迟的单变量对照：
队列、冲击和路由假设 / 未证明项：
结论适用范围 / 保留或更换规则的证据：

季度与策略 ID / 期初计划 / 本期变更：
九项 PnL Attribution / 独立账本 / Residual 调查：
信号、成本、容量、相关和因子的变化：
机制是否仍成立 / 反例 / 失败试验：
风险与纪律事件 / 停机和恢复记录：
策略生命周期决定：维持、降级、暂停、退役：
下季度资本表 / 风险预算 / 研究优先级：
市场结构变化及可验证证据：
提案人 / 风险异议 / 批准人 / 生效时间：

"""Synthetic teaching research, Python 3.10+, standard library only. No orders or network."""
import hashlib
import json
import math
import random
import statistics

SEED = 20260930
HORIZON = 2
LOOKBACK = 10
CANDIDATES = (0.0, 0.003, 0.006)


def validate(returns):
    if len(returns) < 300 or any(not math.isfinite(x) or x <= -1 for x in returns):
        raise ValueError("need >=300 finite simple returns above -1")


def data(seed=SEED):
    rng = random.Random(seed)
    # Independent synthetic returns: this does not assert market predictability.
    return [rng.gauss(0, 0.015) for _ in range(300)]


def rows(returns):
    validate(returns)
    return [dict(t=t, end=t + HORIZON,
                 feature=sum(returns[t-LOOKBACK+1:t+1]),
                 label=math.prod(1+x for x in returns[t+1:t+HORIZON+1])-1)
            for t in range(LOOKBACK-1, len(returns)-HORIZON, HORIZON)]


def segment(samples, start, stop):
    # Purge labels crossing the right boundary; features use only history.
    return [r for r in samples if start <= r["t"] < stop and r["end"] < stop]


def evaluate(samples, threshold, cost_bps=10):
    if not samples or not math.isfinite(cost_bps) or cost_bps < 0:
        raise ValueError("nonempty samples and nonnegative finite cost required")
    if not math.isfinite(threshold) or threshold < 0:
        raise ValueError("threshold must be finite and nonnegative")
    pnl = []
    traded = 0
    for r in samples:
        # Synthetic periods have no opening gap: next open equals t close.
        # Decision after t close, enter next open, then hold two returns.
        position = int(r["feature"] > threshold)
        # Two-way proportional cost; no leverage/financing, no overlapping trades.
        pnl.append(position * r["label"] - 2 * cost_bps / 10000 * position)
        traded += position
    return dict(mean=statistics.mean(pnl), traded=traded, returns=pnl)


def choose(samples):
    # Deterministic tie break: first registered candidate. No test input.
    return max(CANDIDATES, key=lambda p: evaluate(samples, p)["mean"])


def max_drawdown(returns):
    equity = peak = 1.0
    worst = 0.0
    for r in returns:
        if not math.isfinite(r) or r <= -1:
            raise ValueError("invalid path return")
        equity *= 1+r
        peak = max(peak, equity)
        worst = max(worst, 1-equity/peak)
    return worst


def block_bootstrap(returns, block=3, repetitions=1000, seed=SEED):
    if not returns or not isinstance(block, int) or not 1 <= block <= len(returns):
        raise ValueError("invalid block")
    if not isinstance(repetitions, int) or repetitions < 20:
        raise ValueError("need at least 20 repetitions")
    rng = random.Random(seed)
    means, drawdowns = [], []
    for _ in range(repetitions):
        resample = []
        while len(resample) < len(returns):
            start = rng.randrange(len(returns)-block+1)
            resample.extend(returns[start:start+block])
        resample = resample[:len(returns)]
        means.append(statistics.mean(resample))
        drawdowns.append(max_drawdown(resample))
    means.sort()
    drawdowns.sort()
    # Empirical nearest-index percentile, not an analytic confidence guarantee.
    quantile = lambda xs, p: xs[round((len(xs)-1)*p)]
    return dict(mean_interval_95=[quantile(means, .025), quantile(means, .975)],
                drawdown_p50=quantile(drawdowns, .5),
                drawdown_p95=quantile(drawdowns, .95), block=block,
                repetitions=repetitions, seed=seed)


def run(returns):
    samples = rows(returns)
    folds = []
    for boundary in (100, 140, 180):
        training = segment(samples, 0, boundary)
        test = segment(samples, boundary, boundary+40)
        threshold = choose(training)
        folds.append(dict(train_end=boundary, test_end=boundary+40,
                          threshold=threshold, test_mean=evaluate(test, threshold)["mean"]))
    # Development ends at 220. Choose on 180..220 validation; never inspect final test.
    validation = segment(samples, 180, 220)
    selected = choose(validation)
    final = segment(samples, 220, len(returns))
    base = evaluate(final, selected)
    stressed = evaluate(final, selected, cost_bps=30)
    return dict(seed=SEED, input_sha256=hashlib.sha256(json.dumps(returns).encode()).hexdigest(),
                input_rows=len(returns), horizon=HORIZON, attempts=list(CANDIDATES),
                folds=folds, final_start=220, selected=selected, observations=len(final),
                traded=base["traded"], test_net_mean=base["mean"],
                cash_mean=0.0, always_long_round_trip_block_mean=statistics.mean(r["label"] for r in final)-0.002,
                high_cost_mean=stressed["mean"], max_drawdown=max_drawdown(base["returns"]),
                uncertainty=block_bootstrap(base["returns"]))


def self_test():
    original = data()
    samples = rows(original)
    for boundary in (100, 140, 180, 220):
        training = segment(samples, 0, boundary)
        assert all(r["end"] < boundary for r in training)
        assert any(r["t"] < boundary <= r["end"] for r in samples)
    changed = original[:220] + [0.1] * (len(original)-220)
    assert segment(rows(changed), 0, 220) == segment(samples, 0, 220)
    assert run(changed)["selected"] == run(original)["selected"]
    assert all(a["end"] <= b["t"] for a, b in zip(samples, samples[1:]))
    result = run(original)
    assert result == run(original)
    assert result["high_cost_mean"] <= result["test_net_mean"]
    assert abs(max_drawdown([.1, -.2, .1]) - .2) < 1e-12
    for invalid in ([], [0.0]*299, [math.nan]*300, [-1.0]*300):
        try:
            rows(invalid)
        except ValueError:
            pass
        else:
            raise AssertionError("invalid data accepted")
    for block in (0, 1000):
        try:
            block_bootstrap([.01, -.01], block=block)
        except ValueError:
            pass
        else:
            raise AssertionError("invalid block accepted")


if __name__ == "__main__":
    self_test()
    print(json.dumps(run(data()), ensure_ascii=False, indent=2))