第4周跟练：交付一份能复现、讲得清的研究

def timing_rows(prices, cost_bps=5):
    result = backtest(prices, cost_bps)
    rows = []
    for i in range(1, len(prices)):
        # i是收益区间：观测i-1收盘到观测i收盘，不代表真实交易日。
        rows.append(dict(
            interval=i,
            signal_start=i - 3 if i >= 3 else "",
            signal_end=i - 2 if i >= 3 else "",
            execution_observation=i - 1,
            previous_close=prices[i - 1], close=prices[i],
            position=result["positions"][i - 1],
            turnover=result["turnovers"][i - 1],
            net_return=result["returns"][i - 1],
            equity=result["equity"][i - 1],
        ))
    return rows

from week4 import REPORTS, timing_rows, write_csv

# 与独立验收使用不同样例。先在纸上预测，再运行。
prices = [10, 11, 10, 12, 11, 13, 12]
rows = timing_rows(prices, cost_bps=0)
for row in rows:
    print(row)
REPORTS.mkdir(exist_ok=True)
write_csv(REPORTS / "timing-demo.csv", rows)
print("已写入reports/timing-demo.csv。signal_start/end为空表示尚无足够历史。")

"""A reproducible TEACHING experiment, not a live trading strategy.

Only the Python standard library is needed. See README.md for timing,
transaction-cost assumptions, and the six learner tasks.
"""
import csv
import hashlib
import json
import math
from pathlib import Path
import platform
import random

ROOT = Path(__file__).resolve().parent


def validate_prices(prices):
    if len(prices) < 2 or any(not math.isfinite(p) or p <= 0 for p in prices):
        raise ValueError("Need at least two finite, positive prices")


def maximum_drawdown(equity):
    """Positive loss fraction; initial capital of 1 is part of the path."""
    peak, drawdown = 1.0, 0.0
    for value in equity:
        peak = max(peak, value)
        drawdown = max(drawdown, 1 - value / peak)
    return drawdown


def backtest(prices, cost_bps=5):
    validate_prices(prices)
    if not math.isfinite(cost_bps) or not 0 <= cost_bps < 10000:
        raise ValueError("cost_bps must be in [0, 10000)")
    rate = cost_bps / 10000
    previous_position, wealth = 0, 1.0
    positions, returns, equity, turnovers = [], [], [], []
    for i in range(1, len(prices)):
        # Interval i means close[i-1] -> close[i].
        # Decide from the move ending at close[i-2], NOT close[i-1].
        # This leaves a full interval to form the order before execution.
        position = int(prices[i - 2] > prices[i - 3]) if i >= 3 else 0
        turnover = abs(position - previous_position)
        asset_return = prices[i] / prices[i - 1] - 1
        # Simplified proportional fee charged before the interval return.
        growth = (1 - rate * turnover) * (1 + position * asset_return)
        wealth *= growth
        positions.append(position)
        returns.append(growth - 1)
        equity.append(wealth)
        turnovers.append(turnover)
        previous_position = position
    # Mark-to-market at the end; no forced liquidation or exit fee.
    return dict(positions=positions, returns=returns, equity=equity,
                turnovers=turnovers)


def synthetic_prices(seed=2027, observations=301):
    rng = random.Random(seed)
    prices = [100.0]
    for _ in range(observations - 1):
        # Independent lognormal increments: no planted predictive signal.
        prices.append(round(prices[-1] * math.exp(rng.gauss(0.0001, 0.01)), 6))
    return prices


def write_csv(path, rows):
    with path.open("w", encoding="utf-8-sig", newline="") as file:
        writer = csv.DictWriter(file, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)


def main():
    prices = synthetic_prices()
    reports = ROOT / "reports"
    reports.mkdir(exist_ok=True)
    price_rows = [dict(observation=i, close=p) for i, p in enumerate(prices)]
    write_csv(reports / "synthetic_prices.csv", price_rows)
    metrics = []
    for cost in (0, 5, 20):
        result = backtest(prices, cost)
        # Buy-and-hold: one initial purchase, same proportional cost.
        benchmark = [(1 - cost / 10000) * p / prices[0] for p in prices[1:]]
        metrics.append(dict(cost_bps=cost,
                            strategy_return=result["equity"][-1] - 1,
                            strategy_max_drawdown=maximum_drawdown(result["equity"]),
                            benchmark_return=benchmark[-1] - 1,
                            benchmark_max_drawdown=maximum_drawdown(benchmark),
                            total_turnover=sum(result["turnovers"])))
        if cost == 5:
            daily = [dict(interval=i, previous_close=prices[i - 1],
                          close=prices[i], position=result["positions"][i - 1],
                          turnover=result["turnovers"][i - 1],
                          net_return=result["returns"][i - 1],
                          equity=result["equity"][i - 1],
                          benchmark_equity=benchmark[i - 1])
                     for i in range(1, len(prices))]
            write_csv(reports / "daily.csv", daily)
    write_csv(reports / "metrics.csv", metrics)
    manifest = dict(data_kind="synthetic, not historical market data",
                    seed=2027, observations=len(prices),
                    python=platform.python_version(), costs_bps=[0, 5, 20],
                    code_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest(),
                    data_sha256=hashlib.sha256((reports / "synthetic_prices.csv").read_bytes()).hexdigest(),
                    timing="position for interval i uses move ending at close i-2",
                    final_valuation="mark to market, no terminal liquidation cost")
    (reports / "manifest.json").write_text(json.dumps(manifest, indent=2), encoding="utf-8")
    for row in metrics:
        print(f"Cost {row['cost_bps']:2} bp | "
              f"Strategy {row['strategy_return']:.2%} | "
              f"Drawdown {row['strategy_max_drawdown']:.2%} | "
              f"Benchmark {row['benchmark_return']:.2%}")
    print("Teaching output written to reports/. Complete REPORT_TEMPLATE.md.")


if __name__ == "__main__":
    main()

import platform
import sys
import research
from week4 import REPORTS, load_run_prices, timing_rows, write_csv

# 原引擎只用标准库；直接调用原流程，避免另造一套回测公式。
research.main()
prices = load_run_prices()
# 与daily.csv前6行核对；附上信号和成交观测编号。
write_csv(REPORTS / "timing-first-six.csv", timing_rows(prices[:7], 5))
(REPORTS / "environment.txt").write_text(
    f"Python: {platform.python_version()}\nExecutable: {sys.executable}\n"
    f"Platform: {platform.platform()}\nDependencies: Python standard library only\n",
    encoding="utf-8",
)
print("补充完成：timing-first-six.csv、environment.txt。继续运行03_experiments.py。")

def cost_rows(prices):
    rows = []
    base_positions = backtest(prices, 0)["positions"]
    for cost in (0, 5, 20):
        result = backtest(prices, cost)
        # 费用对照必须使用相同数据、信号和持仓路径，只改变费用。
        if result["positions"] != base_positions:
            raise ValueError("持仓路径变了，不是单因素成本对照")
        benchmark = [(1 - cost / 10000) * p / prices[0] for p in prices[1:]]
        rows.append(dict(
            cost_bps=cost,
            strategy_return=result["equity"][-1] - 1,
            strategy_max_drawdown=maximum_drawdown(result["equity"]),
            benchmark_return=benchmark[-1] - 1,
            benchmark_max_drawdown=maximum_drawdown(benchmark),
            total_turnover=sum(result["turnovers"]),
        ))
    if any(b["strategy_return"] > a["strategy_return"] + 1e-12 for a, b in zip(rows, rows[1:])):
        raise ValueError("固定持仓路径下，提高费用不应提高终值")
    return rows


def prefix_check(prices):
    validate_prices(prices)
    if len(prices) < 4:
        raise ValueError("本实验至少需要4个观测")
    changed = list(prices)
    changed[-1] *= 1.5  # 只改最后一个观测；之前的输入完全不动
    before, after = backtest(prices, 5), backtest(changed, 5)
    # 排除受最后一个价格影响的最后收益区间；逐项比较更早结果。
    passed = all(before[key][:-1] == after[key][:-1]
                 for key in ("positions", "returns", "equity", "turnovers"))
    if not passed:
        raise ValueError("修改未来价格影响了此前结果，请检查信息使用时点")
    return dict(changed_observation=len(prices) - 1,
                unchanged_intervals=len(prices) - 2,
                passed=passed,
                limitation="仅检查这一种未来输入变动，不是无前视的完整证明")

import csv
import json
import math
from week4 import ROOT, REPORTS, load_run_prices, cost_rows, prefix_check, result_note, fingerprint, write_csv

prices = load_run_prices()
rows = cost_rows(prices)
# 再核对原引擎metrics，避免辅助脚本与原报告口径悄悄分叉。
with (REPORTS / "metrics.csv").open(encoding="utf-8-sig", newline="") as handle:
    original = list(csv.DictReader(handle))
if len(original) != len(rows) or any(
    not math.isclose(float(original[i][key]), value, rel_tol=1e-12, abs_tol=1e-12)
    for i, row in enumerate(rows) for key, value in row.items()
):
    raise ValueError("成本对照与原metrics不一致，请重跑02并检查代码修改")
write_csv(REPORTS / "experiments.csv", rows)
check = prefix_check(prices)
(REPORTS / "prefix-check.json").write_text(json.dumps(check, ensure_ascii=False, indent=2), encoding="utf-8")
(REPORTS / "result-note.md").write_text(result_note(rows), encoding="utf-8")
# 最后才记录指纹。它只检查文件一致性，不能证明逻辑正确或个人掌握。
(REPORTS / "delivery-manifest.json").write_text(
    json.dumps(fingerprint(ROOT), ensure_ascii=False, indent=2), encoding="utf-8"
)
print(result_note(rows))
print(f"未来变动检查：此前{check['unchanged_intervals']}个区间不变。")
print("已记录delivery-manifest.json。改代码或结果后，请记录原因并重跑02、03。")

def fingerprint(root):
    """对计算代码与reports结果做文件指纹。notes中的人工报告单独人工验收。"""
    paths = list(root.glob("*.py"))
    if (root / "reports").exists():
        paths += [p for p in (root / "reports").iterdir()
                  if p.is_file() and p.name != "delivery-manifest.json"]
    return {p.relative_to(root).as_posix(): sha256(p) for p in sorted(paths)}


def verify_fingerprint(root, expected):
    actual = fingerprint(root)
    if actual != expected:
        changed = sorted(name for name in set(actual) | set(expected)
                         if actual.get(name) != expected.get(name))
        raise ValueError("文件已变化、缺失或新增：" + ", ".join(changed))
    return len(actual)

import json
from week4 import ROOT, REPORTS, verify_fingerprint

manifest = REPORTS / "delivery-manifest.json"
if not manifest.exists():
    raise SystemExit("先依次完成01、02、03，生成交付指纹。")
count = verify_fingerprint(ROOT, json.loads(manifest.read_text(encoding="utf-8")))
print(f"计算文件与结果指纹一致：{count}个文件。")
print("这不是报告评分：请人工检查notes/report.md、失败记录、贡献说明与5分钟讲解。")
print("还需在新文件夹重跑10项测试与01–04；指纹一致不能代替重跑。")

def result_note(rows):
    lines = [
        "# 自动生成的结果核对单（不是个人研究报告）", "",
        "来源：本次合成价格；同一策略持仓路径，费用为单边仓位变化费用。",
        "观测编号不是真实交易日，不年化；未模拟期末强制平仓。", "",
        "|费用bp|策略累计收益|策略最大回撤|基准累计收益|基准最大回撤|累计仓位变化|",
        "|---|---|---|---|---|---|",
    ]
    for r in rows:
        lines.append(f'|{r["cost_bps"]}|{r["strategy_return"]:.4%}|'
                     f'{r["strategy_max_drawdown"]:.4%}|{r["benchmark_return"]:.4%}|'
                     f'{r["benchmark_max_drawdown"]:.4%}|{r["total_turnover"]}|')
    row = next(r for r in rows if r["cost_bps"] == 5)
    difference = 100 * (row["strategy_return"] - row["benchmark_return"])
    lines += ["", f"5bp下，策略累计收益减基准累计收益为{difference:+.4f}个百分点。",
              "这是两个累计收益的算术差，不是回归alpha，也不是风险调整后的超额收益。",
              "该结果不能证明真实市场有效；费用对照只检验简化成本模型中的敏感性。",
              "", "请自己补充：问题、时点假设、具体证据、失败、至少3项局限、个人贡献与下一步。",
              "不要删除负结果，不要把本核对单当成自己已完成的报告。"]
    return "\n".join(lines) + "\n"

# 研究报告：[自己填写标题]

> 目标1–2页正文，明细和日志另附。把本文件复制到notes/report.md再填写。
> 所有[]必须换成你实际验证的内容；不知道就写“未验证”及补查办法。

## 1. 问题与一句话结论
问题：[固定信号在三种费用下怎样变化？什么现象会让你怀疑实现有错？]
结论：[在什么数据与假设下，看到了什么；不能外推到哪里？]

## 2. 数据与范围
- 来源：本站原创合成生成器；seed=[ ]，观测数=[ ]，区间数=[ ]。
- 观测编号是顺序标签，不是交易日；不是A股行情，没有历史下载日期。
- 快照位置：reports/synthetic_prices.csv；SHA256见manifest.json。
- 检查：[完整正价格、时点手算等实际检查，注明文件/行或观测编号]。

## 3. 方法与基准
- 区间i=[ ]；信号观察到=[ ]；执行假设=[ ]；持仓作用区间=[ ]。
- 仓位定义、换手定义、费用扣除公式：[ ]。
- 基准：[买入一次、同费用口径、期末估值方式]。
- 未建训练模型，未宣称独立样本外验证；若尝试其他规则，完整列出尝试。

## 4. 所有成本结果
从本次reports/experiments.csv抄写，勿凭记忆填写。

|费用bp|策略累计收益|策略最大回撤|基准累计收益|基准最大回撤|累计仓位变化|
|---|---|---|---|---|---|
|0|[ ]|[ ]|[ ]|[ ]|[ ]|
|5|[ ]|[ ]|[ ]|[ ]|[ ]|
|20|[ ]|[ ]|[ ]|[ ]|[ ]|

解释：[先写观察，再写可能原因，最后写尚不能证明什么。]
累计收益差用“百分点”；不是alpha。不把观测数直接当交易日进行年化。

## 5. 失败与局限
至少3条，每条写“限制→影响→如何核查或改进”。
- 数据：[ ]。
- 成交/成本：[ ]。
- 统计证据/信息时点：[ ]。
- 失败实验：[预期、实际、根因、修正、复测，指向notes/experiment-log.csv]。

## 6. 个人贡献与AI辅助
- 模板已提供：[引擎、信号、生成器、测试或辅助脚本等]。
- 我独立完成：[具体文件/解释/实验，以及如何证明]。
- AI参与：[生成、解释或调试的具体内容，我怎样核验]。
- 我仍不能独立解释：[ ]。

## 7. 下一步：一个可检验的小任务
假设=[ ]；只改=[ ]；保持=[ ]；预期区分的两种解释=[ ]；
成功/失败判定=[ ]；预计用时=[ ]。不能只写“多调参”。

## 8. 复现
Python版本、执行顺序、测试数量与实际结果：[ ]。
本次引擎/数据哈希见manifest.json；全部计算文件指纹见delivery-manifest.json。
新目录重跑日期、目录、差异说明：[ ]。个人报告与日志另人工核对，不由指纹自动评分。

# 5分钟讲解（不照念完整报告）

- 0:00–0:40：研究问题是什么，数据是不是市场数据？
- 0:40–1:40：用一个区间解释可得信息、信号、执行、收益；指向timing-first-six.csv。
- 1:40–2:40：三种费用全部结果，基准是什么；收益和回撤怎样计算？
- 2:40–3:40：一次失败/检查，以及至少3个具体局限。
- 3:40–4:30：模板做了什么、你做了什么、AI做了什么，展示一处证据。
- 4:30–5:00：一个下一步实验，只改什么，怎样判断结果。

模拟追问：
1. 为什么不能说“代码shift了，所以没有未来信息”？
2. 原始结果为负，报告还要交吗？
3. 换一台机器能复现吗？哈希相同是否证明策略可靠？
4. 这段程序能直接用于A股吗？缺了哪些具体规则？
5. 去掉AI，你能现场解释哪一行，独立修改哪个小实验？

# 第4周交付验收（自评，不是自动阅卷）

## 文件
- [ ] 根目录保留研究引擎、辅助脚本、两个test文件与README。
- [ ] reports中保留本次数据快照、逐期/成本结果、时点表、两个manifest、环境与前缀检查。
- [ ] notes/report.md是我填写的报告；experiment-log.csv包含全部实验和失败；有手算与讲解提纲。
- [ ] 所有数值来自本次输出，原始模板与我的修改可区分。
- [ ] 交付压缩包不含.venv、__pycache__、密钥、账户信息或未授权的数据。

## 新目录复现
- [ ] 保留交付快照，另建空目录，复制代码/模板/notes，重新建环境，不复制旧reports。
- [ ] 运行10项测试，再依次运行01–04，全部成功退出。
- [ ] 比较旧/新synthetic_prices.csv、daily.csv、metrics.csv、experiments.csv。
- [ ] Python或平台变化时记录environment.txt及manifest差异，不能把忽略差异当作通过。
- [ ] 04检查只验证计算文件完整性，不验证报告逻辑或我是否独立完成。

## 60分钟独立验收：5项各0–2分
0=不能完成；1=需提示或只能复述；2=独立完成并解释。

|项目|分数|证据|
|---|---|---|
|时间边界（必须2分）|||
|计算与成本（必须2分）|||
|可复现交付|||
|研究表达|||
|独立贡献|||

合计至少8/10，关键项必须各2分。时间安排：60分钟验收、15分钟评分、25分钟修正、20分钟下月安排。

隔24–48小时：只拿价格[100,120,100,110,100,110]，费用0，20分钟手算5段持仓/收益/累计值；
从空白提纲讲清假设、最薄弱证据和下一步，再运行代码核对。不要提前看网页折叠答案。