#!/usr/bin/env python3 """Все числа TW-012 одним прогоном. Урок: «после консолидации начинается сильное движение». Проверка. ЗАЧЕМ ЭТОТ ФАЙЛ ЛЕЖИТ В ПРОЕКТЕ. Канал обещает: всё измерено, а не заявлено. Обещание стоит ровно столько, сколько стоит возможность пересчитать. Числа TW-012 считались до того, как правило завелось, и жили только текстом в шапке сценария — проверить их было не по чему. python consolidation.py # всё python consolidation.py # сам пишет results.txt (G-18) ЧТО ГДЕ ЗВУЧИТ: s06-s07 SPY, двенадцать сессий 28.11 — 15.12: полоса 10.51 при среднем ходе 5.32, отношение 1.98 s10 сколько всего окон, медиана, граница тесной четверти, максимум s11 пороги: под 3 консолидация, около 3.5 обычно, выше 4 тренд s14 тесная четверть против остальных на следующих десяти сессиях: покрытый диапазон и направленное движение МЕТОД. Окно — двенадцать подряд сессий. Полоса окна — самый высокий максимум минус самый низкий минимум. Ширина окна — полоса, делённая на средний дневной ход ЭТИХ ЖЕ двенадцати дней: так число говорит «столько дневных ходов уместилось в полосу», и тесная неделя отличается от широкой, а не дорогой инструмент от дешёвого. ПОЧЕМУ ПОСЛЕДСТВИЯ НОРМИРУЮТСЯ ИНАЧЕ. Диапазон, покрытый за следующие десять сессий, делится на средний дневной ход ВСЕЙ выборки инструмента, а не самого окна. Делить на ход окна нельзя: у тесного окна он мал по определению, и деление на него завышает результат именно у той группы, которую и проверяем. Так получается разница из ниоткуда — 2.87 против 3.22 вместо честных 3.14 против 3.11. Это ровно та ошибка, о которой ролик и говорит: сравнивать надо с чем-то, что от проверяемого признака не зависит. ДАННЫЕ. `04_assets/market/*-1d.json`, шесть инструментов. Берутся последние 200 сессий каждого: у акций это 9,5 месяца, у крипты 6,5 (s15). """ from __future__ import annotations import io import json import statistics as st import sys from pathlib import Path STUDIO = Path(__file__).resolve().parents[3] MARKET = STUDIO / "tickwright" / "04_assets" / "market" NAMES = ["AAPL", "TSLA", "SPY", "BTCUSDT", "ETHUSDT", "SOLUSDT"] WIN = 12 # сессий в окне AFTER = 10 # сессий после окна TAIL = 200 # сессий на инструмент (акции 9,5 мес., крипта 6,5 мес.) def bars(name: str) -> list[dict]: d = json.loads((MARKET / f"{name}-1d.json").read_text(encoding="utf-8")) return d["bars"] if isinstance(d, dict) and "bars" in d else d def main() -> int: for s in (sys.stdout, sys.stderr): if hasattr(s, "reconfigure"): s.reconfigure(encoding="utf-8", errors="replace") rows_by = {n: bars(n)[-TAIL:] for n in NAMES} print("ДАННЫЕ") for n, r in rows_by.items(): print(f" {n:8} {len(r)} сессий {r[0]['d']} … {r[-1]['d']}") # средний дневной ход инструмента за всю его выборку whole = {n: sum(b["h"] - b["l"] for b in r) / len(r) for n, r in rows_by.items()} widths, follow = [], [] for n, rows in rows_by.items(): for i in range(len(rows) - WIN + 1): w = rows[i:i + WIN] lo, hi = min(b["l"] for b in w), max(b["h"] for b in w) band = hi - lo adr = sum(b["h"] - b["l"] for b in w) / WIN if adr <= 0: continue width = band / adr # низ и верх — в хвосте кортежа, чтобы прежние индексы не сдвинулись widths.append((n, rows[i]["d"], rows[i + WIN - 1]["d"], width, band, adr, lo, hi)) if i + WIN + AFTER <= len(rows): a = rows[i + WIN:i + WIN + AFTER] cov = max(b["h"] for b in a) - min(b["l"] for b in a) direc = abs(a[-1]["c"] - w[-1]["c"]) follow.append((width, cov / whole[n], direc / whole[n])) ws = [x[3] for x in widths] q1 = st.quantiles(ws, n=4)[0] print("\ns06-s07 · пример на SPY") spy = [x for x in widths if x[0] == "SPY"] tight_spy = min(spy, key=lambda x: x[3]) print(f" самое тесное окно SPY: {tight_spy[1]} … {tight_spy[2]}") # НИЗ И ВЕРХ ПОЛОСЫ ПЕЧАТАЮТСЯ: голос s07 называет их («from six # hundred and seventy nine to six hundred and eighty nine»), а вывод # до 8 октября давал только ширину — цены в речи сверять было не с чем. print(f" низ {tight_spy[6]:.2f} · верх {tight_spy[7]:.2f}") print(f" полоса {tight_spy[4]:.2f} · средний ход {tight_spy[5]:.2f} " f"· отношение {tight_spy[3]:.2f}") print(" (в ролике: 28.11 — 15.12, полоса 10.51, ход 5.32, 1.98)") print("\ns10 · как выглядит обычное") print(f" окон всего: {len(ws)}") print(f" медиана: {st.median(ws):.2f}") print(f" граница самой тесной четверти: {q1:.2f}") widest = max(widths, key=lambda x: x[3]) # Два знака, как в описании: расчёт обязан быть не грубее текста, # который его цитирует, иначе сверить цитату не с чем. print(f" самое широкое: {widest[3]:.2f} — {widest[0]}, " f"{widest[1]} … {widest[2]}") print(" (в ролике: 1 128 окон, медиана 3.5, четверть под 3, " "максимум 10.1)") print("\ns11 · пороги") for lo, hi, what in ((0, 3.0, "консолидация"), (3.0, 4.0, "обычный"), (4.0, 99, "тренд")): share = sum(1 for w in ws if lo <= w < hi) / len(ws) * 100 print(f" {what:14} {share:5.1f}% окон") print("\ns14 · что было потом") tight = [f for f in follow if f[0] <= q1] rest = [f for f in follow if f[0] > q1] print(f" окон с продолжением: {len(follow)} " f"(тесных {len(tight)}, остальных {len(rest)})") print(f" покрытый диапазон: {st.median([f[1] for f in tight]):.2f} " f"после тесных · {st.median([f[1] for f in rest]):.2f} " f"после обычных") print(f" направленное: {st.median([f[2] for f in tight]):.2f} " f"после тесных · {st.median([f[2] for f in rest]):.2f} " f"после обычных") print(" (в ролике: 3.09 против 3.14 · 1.23 против 1.41)") print("\nВЫВОД. Сильного движения после тесного окна нет: покрытый " "диапазон\nодинаков, а направленное движение у тесных окон даже " "меньше. Это то же,\nчто говорит ролик.") 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())