#!/usr/bin/env python3 """Все числа TW-013 одним прогоном. Расхожее утверждение: «уровень, образованный гепом, является очень сильным». Проверка. ЗАЧЕМ ЭТОТ ФАЙЛ ЛЕЖИТ В ПРОЕКТЕ. Канал обещает: всё измерено, а не заявлено. Обещание стоит ровно столько, сколько стоит возможность пересчитать. До TW-013 числа роликов жили только в шапке сценария — проверить их было не по чему. Теперь расчёт лежит рядом с кадрами, читает те же котировки, что и кадр-график, и печатает ровно те числа, которые звучат голосом. python gaps.py # всё python gaps.py # сам пишет results.txt (G-18) ЧТО ГДЕ ЗВУЧИТ: s02 37 % сессий открываются вне вчерашнего диапазона s03 медиана гепа 0.22 дневного хода; три четверти под 0.50; порог 0.50 → 8 % сессий, 476 событий за восемь лет s09 476 гепов, 3 инструмента, 8 лет; плацебо — 448 s12 доля отхода назад: 0.47 / 0.52 / 0.45 против 0.49 / 0.53 / 0.50 s13 встал и закрылся обратно в день касания: 27 / 27 / 24 % против 32 / 32 / 28 % s14 край пустой зоны: 0.50 и контроль 0.50 s15 без первого дня, с равного расстояния (1.2 дневного хода): медиана 6 сессий против 9; за 60 сессий 79 % против 71 % s17 медиана прохода за уровень 1.4 дневного хода по 435 замерам МЕТОД. Событие — геп не меньше половины среднего дневного хода за прошлые 20 сессий. Контроль — день, прошедший столько же (в пределах 0.15 дневного хода), но открывшийся ВНУТРИ вчерашнего диапазона. Контроль именно такой, а не линия наугад: случайную линию цена трогает через двадцать три сессии в спокойном рынке, а край гепа — через две, сразу после шока. Сравнивая их, меряешь погоду, а не уровень (rules.md §9, 6 сентября). ДАННЫЕ. `04_assets/market/{AAPL,TSLA,SPY}-1d.json`, сняты 6 сентября 2026, дневные бары с 20 сентября 2018 по 4 сентября 2026. Крипта не участвует: торгуется круглосуточно, гепов не образует. """ from __future__ import annotations import io import json import random import sys from pathlib import Path from statistics import median STORE = Path(__file__).resolve().parents[1] / "market" SYMS = ["AAPL", "TSLA", "SPY"] ADR_N = 20 # окно среднего дневного хода — как в TW-010 и TW-012 MIN_GAP = 0.5 # геп меньше половины дневного хода событием не считаем AFTER = 5 # сколько сессий смотрим после первого касания TOL = 0.15 # насколько похожим должен быть ход дня у плацебо STOP = 0.25 # «встал»: прошёл меньше четверти дневного хода за уровень SEED = 17 # номер урока; фиксирован, чтобы счёт повторялся дословно def load(sym: str) -> list[dict]: return json.load(io.open(STORE / f"{sym}-1d.json", encoding="utf-8"))["bars"] BARS = {s: load(s) for s in SYMS} def adr(bars: list[dict], i: int, n: int = ADR_N) -> float: """Средний дневной ход за n сессий ДО i. Единица измерения всего канала.""" w = bars[max(0, i - n):i] return sum(b["h"] - b["l"] for b in w) / len(w) if w else 0.0 def touch(bars: list[dict], start: int, level: float) -> int | None: """Первая сессия с start, чей диапазон накрывает уровень.""" for j in range(start, len(bars)): if bars[j]["l"] <= level <= bars[j]["h"]: return j return None def points(bars: list[dict], i: int) -> dict: """Три цены урока плюс край пустой зоны, который урок не называет.""" p, b = bars[i - 1], bars[i] up = b["c"] >= p["c"] return {"prev_close": p["c"], "open": b["o"], "extreme": b["h"] if up else b["l"], "edge": p["h"] if up else p["l"]} def events() -> tuple[list[dict], list[dict], list[float], int]: """События, плацебо к ним, размеры всех гепов и число сессий.""" random.seed(SEED) gaps, plain, sizes, sessions = [], [], [], 0 for sym in SYMS: bars = BARS[sym] sessions += len(bars) - ADR_N for i in range(ADR_N, len(bars) - AFTER - 2): p, b = bars[i - 1], bars[i] a = adr(bars, i) if a <= 0: continue rec = {"sym": sym, "i": i, "d": b["d"], "adr": a, "move": (b["c"] - p["c"]) / a, "pts": points(bars, i)} if b["o"] > p["h"] or b["o"] < p["l"]: size = (b["o"] - p["h"]) if b["o"] > p["h"] else (p["l"] - b["o"]) sizes.append(size / a) if size / a >= MIN_GAP: rec["size"] = size / a gaps.append(rec) else: plain.append(rec) pool: dict[str, list[dict]] = {} for r in plain: pool.setdefault(r["sym"], []).append(r) placebo = [] for g in gaps: near = [r for r in pool[g["sym"]] if abs(r["move"] - g["move"]) <= TOL] if near: placebo.append(random.choice(near)) return gaps, placebo, sorted(sizes), sessions def react(bars: list[dict], j: int, level: float) -> tuple[float, float] | None: """Доля отхода назад и проход за уровень за AFTER сессий после касания. Направление подхода — по закрытию предыдущей сессии. 0,50 значит, что уровень не значит ничего: цена ушла за него ровно настолько же, насколько отошла назад. """ a = adr(bars, j) if a <= 0 or j == 0 or j + AFTER >= len(bars): return None w = bars[j:j + AFTER + 1] hi, lo = max(b["h"] for b in w), min(b["l"] for b in w) from_below = bars[j - 1]["c"] < level through = max(0.0, (hi - level) if from_below else (level - lo)) back = max(0.0, (level - lo) if from_below else (hi - level)) return (back / (through + back), through / a) if through + back > 0 else None def five_sessions(cases: list[dict], key: str) -> tuple[int, float, float]: vals, past = [], [] for c in cases: bars = BARS[c["sym"]] j = touch(bars, c["i"] + 1, c["pts"][key]) r = react(bars, j, c["pts"][key]) if j is not None else None if r: vals.append(r[0]) past.append(r[1]) return len(vals), median(vals), median(past) def touch_day(cases: list[dict], key: str) -> tuple[int, float]: """Встала ли цена в САМ день касания — версия, честная к «локальной остановке»: мера за пять сессий мерит уже следующий ход, а не остановку.""" n = stops = 0 for c in cases: bars, lv = BARS[c["sym"]], c["pts"][key] j = touch(bars, c["i"] + 1, lv) if j is None or j == 0: continue a = adr(bars, j) if a <= 0: continue from_below = bars[j - 1]["c"] < lv past = (bars[j]["h"] - lv) if from_below else (lv - bars[j]["l"]) back = bars[j]["c"] < lv if from_below else bars[j]["c"] > lv n += 1 stops += (max(0.0, past) / a < STOP) and back return n, stops / n * 100 def random_lines(gaps: list[dict], per: int = 3) -> tuple[int, float, float, float]: """Контроль, от которого мы ОТКАЗАЛИСЬ, — и числа, объясняющие почему. Линия наугад внутри диапазона предыдущих 60 сессий. Считаем, через сколько сессий её трогают и насколько широко ходит инструмент в этот момент. Если ждать приходится вчетверо дольше и в вечно спокойном рынке, то сравнение с краем гепа меряет не уровень, а погоду (сцена 10). """ random.seed(SEED + 1) waits, vol = [], [] for e in gaps: bars = BARS[e["sym"]] w = bars[max(0, e["i"] - 60):e["i"]] lo, hi = min(b["l"] for b in w), max(b["h"] for b in w) for _ in range(per): j = touch(bars, e["i"] + 1, random.uniform(lo, hi)) if j is None or adr(bars, j) <= 0: continue waits.append(j - e["i"]) vol.append(adr(bars, j) / e["adr"]) return len(waits), median(waits), median(vol), 0.0 def vol_at_touch(cases: list[dict], key: str) -> float: """Во сколько раз шире ходит инструмент в день касания уровня.""" out = [] for c in cases: bars = BARS[c["sym"]] j = touch(bars, c["i"] + 1, c["pts"][key]) if j is None or adr(bars, j) <= 0: continue out.append(adr(bars, j) / c["adr"]) return median(out) def survivors(cases: list[dict]) -> list[dict]: """Не закрывшиеся в первый день. Без этого условия сравнение — арифметика: день без гепа открылся внутри вчерашнего диапазона и стартует ближе.""" return [c for c in cases if not (BARS[c["sym"]][c["i"]]["l"] <= c["pts"]["prev_close"] <= BARS[c["sym"]][c["i"]]["h"])] def comeback(cases: list[dict]) -> tuple[float, list[int], dict[int, float]]: dists, waits, hit = [], [], {5: 0, 20: 0, 60: 0} seen = {5: 0, 20: 0, 60: 0} for c in cases: bars, target = BARS[c["sym"]], c["pts"]["prev_close"] dists.append(abs(bars[c["i"]]["c"] - target) / c["adr"]) j = touch(bars, c["i"] + 1, target) if j is not None: waits.append(j - c["i"]) for h in hit: if c["i"] + 1 + h < len(bars): seen[h] += 1 hit[h] += any(b["l"] <= target <= b["h"] for b in bars[c["i"] + 1:c["i"] + 1 + h]) return median(dists), sorted(waits), {h: hit[h] / seen[h] * 100 for h in hit} def main() -> int: sys.stdout.reconfigure(encoding="utf-8", errors="replace") gaps, placebo, sizes, sessions = events() print("ДАННЫЕ") for s in SYMS: b = BARS[s] print(f" {s:<5} {len(b)} баров {b[0]['d']} … {b[-1]['d']}") print(f" сессий в счёте: {sessions}\n") print("s02 · s03 — насколько геп обычен") print(f" гепов любого размера: {len(sizes)} " f"({len(sizes) / sessions * 100:.0f}% сессий)") print(f" медиана размера: {median(sizes):.2f} дневного хода") print(f" три четверти меньше: {sizes[int(len(sizes) * .75)]:.2f}") print(f" событий (от {MIN_GAP}): {len(gaps)} " f"({len(gaps) / sessions * 100:.1f}% сессий)\n") print(f"s09 — выборка и плацебо") print(f" событий {len(gaps)}, плацебо подобрано {len(placebo)}") print(f" ход дня: геп {median([g['move'] for g in gaps]):+.2f}, " f"плацебо {median([p['move'] for p in placebo]):+.2f} дневного хода\n") print("s10 — почему контроль не «линия наугад»") n, w, v, _ = random_lines(gaps) print(f" линия наугад: ждать {w:.0f} сессий, " f"дневной ход в день касания ×{v:.2f} (n={n})") for name, cases in (("геп", gaps), ("плацебо", placebo)): for key, label in (("open", "открытие"), ("extreme", "экстремум")): ws = [touch(BARS[c['sym']], c['i'] + 1, c['pts'][key]) for c in cases] ws = [j - c["i"] for j, c in zip(ws, cases) if j is not None] print(f" {name} · {label}: ждать {median(ws):.0f} сессии, " f"дневной ход ×{vol_at_touch(cases, key):.2f}") # Дневной ход в день касания у линии наугад и у края гепа почти одинаков: # средний ход за 20 сессий один геповый бар почти не двигает. Значит # разница между этими контролями — В СРОКЕ И В ОБСТАНОВКЕ, а не в широте # хода. Фраза «инструмент ходит вдвое шире» стояла в первой версии сцены # 10 и снята этим замером до озвучки. print() print(f"s12 · s14 — доля отхода назад за {AFTER} сессий (0.50 — ничего)") names = {"prev_close": "1 · закрытие перед событием", "open": "2 · открытие дня", "extreme": "3 · экстремум дня", "edge": "· край пустой зоны (урок его не называет)"} for key, label in names.items(): n1, r1, t1 = five_sessions(gaps, key) n2, r2, _ = five_sessions(placebo, key) print(f" {label:<42} геп {r1:.2f} (n={n1}, проход {t1:.2f}) " f"плацебо {r2:.2f} (n={n2})") print(f"\ns13 — встал и закрылся обратно в день касания") for key, label in names.items(): n1, s1 = touch_day(gaps, key) n2, s2 = touch_day(placebo, key) print(f" {label:<42} геп {s1:4.1f}% (n={n1}) плацебо {s2:4.1f}% (n={n2})") print(f"\ns15 — возврат к старой цене, БЕЗ первого дня") for name, cases in (("геп", gaps), ("плацебо", placebo)): surv = survivors(cases) dist, waits, rates = comeback(surv) share = len(surv) / len(cases) * 100 print(f" {name:<8} не закрылись в первый день: {len(surv)} ({share:.0f}%)") print(f" расстояние до цели на конец дня: {dist:.2f} дневного хода") print(f" дальше ждать: медиана {median(waits):.0f} сессий") print(f" закрылось за 5/20/60 сессий: " f"{rates[5]:.0f}% / {rates[20]:.0f}% / {rates[60]:.0f}%") print(f"\ns08 · s17 — пример в кадре: AAPL, 5 августа 2024") bars = BARS["AAPL"] k = next(i for i, b in enumerate(bars) if b["d"] == "2024-08-05") p, b = bars[k - 1], bars[k] a = adr(bars, k) print(f" закрытие перед гепом {p['c']:.2f}, минимум дня перед гепом {p['l']:.2f}") print(f" открытие {b['o']:.2f}, минимум дня {b['l']:.2f}") print(f" пустая зона {p['l'] - b['o']:.2f} при дневном ходе {a:.2f} " f"= {(p['l'] - b['o']) / a:.2f} дневного хода") j = touch(bars, k + 1, p["c"]) w = bars[j:j + AFTER + 1] hi = max(x["h"] for x in w) print(f" возврат к {p['c']:.2f}: {bars[j]['d']}, через {j - k} сессий") print(f" дальше {hi:.2f} — на {hi - p['c']:.2f} за уровень, " f"{(hi - p['c']) / adr(bars, j):.2f} дневного хода") return 0 def _run() -> int: """Прогон со снимком: что напечатано, то и легло в `results.txt`. Раньше снимок писался перенаправлением (`python x.py > results.txt`), и прогон без перенаправления оставлял прежний файл: сравнение «снимок совпал» сравнивало копию саму с собой (G-18 второго аудита). Теперь — как у остальных этюдов курса. Снимок пишется только при удачном прогоне: обрубок опаснее отсутствия, он выглядит как настоящий отчёт. """ real = sys.stdout for s in (real, sys.stderr): if hasattr(s, "reconfigure"): s.reconfigure(encoding="utf-8", errors="replace") buf = io.StringIO() class _Tee: """Поток, который пишет в два места и не спорит с вызовами скрипта.""" encoding = "utf-8" errors = "replace" def write(self, text): real.write(text) buf.write(text) return len(text) def flush(self): real.flush() def reconfigure(self, **kw): pass def isatty(self): return False sys.stdout = _Tee() try: code = main() finally: sys.stdout = real if code == 0: out = Path(__file__).with_name("results.txt") out.write_text(buf.getvalue(), encoding="utf-8", newline="\n") print(f"\nснимок: {out}") return code if __name__ == "__main__": raise SystemExit(_run())