#!/usr/bin/env python3 """Все числа TW-021 одним прогоном. Тема R-04: канал как измерение. ЗАЧЕМ ЭТОТ ФАЙЛ ЛЕЖИТ В ПРОЕКТЕ. Канал обещает: всё измерено, а не заявлено. Обещание стоит ровно столько, сколько стоит возможность пересчитать. ЧТО ПРОВЕРЯЕТСЯ. Канал рисуют по двум-трём точкам на глаз, и он выглядит убедительно всегда — потому что через любые точки можно провести линию. Здесь канал СЧИТАЕТСЯ: прямая по методу наименьших квадратов плюс полоса в два стандартных отклонения. И дальше проверяется то, что от канала ждут. 1. Сколько времени цена проводит внутри полосы. Если «внутри» — почти всегда, канал не сообщает ничего: он описывает, а не предсказывает. 2. Что происходит ПОСЛЕ касания границы. Расхожее чтение: от границы цена возвращается к середине. Считается доля возвратов и сравнивается с базой — тем, что делает цена вообще, без всякого канала. 3. Насколько граница устойчива. Канал перерисовывают каждый день; если от одного нового бара наклон прыгает, то «граница» — это не уровень, а сегодняшнее мнение регрессии. 4. Насколько ответ зависит от длины окна. Двадцать дней, шестьдесят и сто двадцать дают три разных канала на одном и том же графике — и это главный довод против «канал очевиден». ПОЧЕМУ ИМЕННО РЕГРЕССИЯ. Это самый простой способ провести линию так, чтобы результат не зависел от того, кто её проводит. Рисование по касаниям даёт столько каналов, сколько людей, и спорить о них невозможно; регрессия даёт один и тот же ответ у всех, и его можно проверить. ДАННЫЕ. `04_assets/market/*-1d.json`, шесть инструментов целиком. """ from __future__ import annotations 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"] WINDOWS = [20, 60, 120] BAND = 2.0 # ширина полосы в стандартных отклонениях AHEAD = 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 fit(closes: list[float]) -> tuple[float, float, float]: """Прямая по наименьшим квадратам и разброс вокруг неё. Возвращает наклон, свободный член и стандартное отклонение остатков. Наклон — в единицах цены за день; чтобы сравнивать инструменты, ниже он всегда делится на цену. """ n = len(closes) xs = list(range(n)) mx, my = sum(xs) / n, sum(closes) / n num = sum((x - mx) * (y - my) for x, y in zip(xs, closes)) den = sum((x - mx) ** 2 for x in xs) or 1e-9 slope = num / den base = my - slope * mx resid = [y - (slope * x + base) for x, y in zip(xs, closes)] return slope, base, st.pstdev(resid) or 1e-9 def inside(r: list[dict], w: int) -> float: """Доля дней, где закрытие попало внутрь полосы своего канала.""" got = hit = 0 for i in range(w, len(r)): c = [b["c"] for b in r[i - w:i]] slope, base, sd = fit(c) mid = slope * w + base got += 1 if abs(r[i]["c"] - mid) <= BAND * sd: hit += 1 return hit / got * 100 if got else 0.0 def touch_outcome(r: list[dict], w: int) -> dict: """Что было после касания границы — против того, что бывает вообще. ВОПРОС ОДИН И ТОТ ЖЕ, и в этом всё дело. «Цена через пять дней пошла в сторону середины» — для касания верхней границы это значит «ниже, чем сегодня», для нижней — «выше». База считает ровно тот же вопрос из любого дня. ПЕРВАЯ ВЕРСИЯ ЭТОЙ ФУНКЦИИ СРАВНИВАЛА РАЗНОЕ. Она требовала от касания дойти до средней линии — то есть пройти два стандартных отклонения, — а от обычного дня хватало пересечь середину с расстояния в полшага. Числа выходили втрое хуже базы, и «вывод» был бы артефактом мерки, а не свойством рынка. """ back = touch = base_back = base_all = 0 for i in range(w, len(r) - AHEAD): c = [b["c"] for b in r[i - w:i]] slope, b0, sd = fit(c) mid = slope * w + b0 upper, lower = mid + BAND * sd, mid - BAND * sd price, later = r[i]["c"], r[i + AHEAD]["c"] base_all += 1 if (price > mid and later < price) or (price < mid and later > price): base_back += 1 if price >= upper or price <= lower: touch += 1 if (price >= upper and later < price) or (price <= lower and later > price): back += 1 return {"touch": touch, "back": back / touch * 100 if touch else 0.0, "base": base_back / base_all * 100 if base_all else 0.0} def slope_jitter(r: list[dict], w: int) -> float: """На сколько процентов наклон меняется от ОДНОГО нового бара. Берётся медиана относительного изменения. Это и есть ответ на вопрос, можно ли называть границу канала уровнем: уровень стоит на месте, а эта линия каждый день немного другая. """ out = [] prev = None for i in range(w, len(r)): slope, _, _ = fit([b["c"] for b in r[i - w:i]]) rel = slope / r[i]["c"] * 100 * 20 # % за 20 дней if prev is not None: out.append(abs(rel - prev)) prev = rel return st.median(out) if out else 0.0 def window_spread(r: list[dict], at: int) -> dict: """Три окна на одном и том же дне — три разных канала. Возвращает наклоны и полуширины в процентах цены. Числа считаются на последнем дне ряда: это тот самый график, на который смотрит зритель. """ out = {} for w in WINDOWS: c = [b["c"] for b in r[at - w:at]] slope, base, sd = fit(c) price = r[at - 1]["c"] out[w] = {"slope": slope / price * 100 * 20, # % за 20 дней "half": BAND * sd / price * 100} return out def main() -> int: for s in (sys.stdout, sys.stderr): if hasattr(s, "reconfigure"): s.reconfigure(encoding="utf-8", errors="replace") rows = {n: bars(n) for n in NAMES} print("ДАННЫЕ") for n, r in rows.items(): print(f" {n:9} {len(r)} баров {r[0]['d']} … {r[-1]['d']}") print(f"\n1 · СКОЛЬКО ВРЕМЕНИ ЦЕНА ВНУТРИ ПОЛОСЫ (±{BAND:.0f} σ)") print(f" {'':11}" + "".join(f"{str(w) + ' дн':>10}" for w in WINDOWS)) for n, r in rows.items(): print(f" {n:9} " + "".join(f"{inside(r, w):9.1f}%" for w in WINDOWS)) print(" Полоса описывает прошлое почти целиком — и это её предел:") print(" то, что охватывает почти всё, ничего не выделяет.") print(f"\n2 · ПОСЛЕ КАСАНИЯ — ПОШЛА ЛИ ЦЕНА К СЕРЕДИНЕ ЗА {AHEAD} ДНЕЙ") print(f" {'':11}{'касаний':>9}{'к середине':>12}{'база':>9}" f"{'разница':>10}") for n, r in rows.items(): o = touch_outcome(r, 60) print(f" {n:9} {o['touch']:8} {o['back']:11.1f}% " f"{o['base']:7.1f}% {o['back'] - o['base']:+9.1f}") print(" (окно 60 дней · база — то же событие из любого дня, не только") print(" из дня касания)") print("\n3 · НАСКОЛЬКО УСТОЙЧИВА ГРАНИЦА") print(" Медианное изменение наклона от ОДНОГО нового бара,") print(" в процентных пунктах наклона за 20 дней — в тех же") print(" единицах, в каких наклон показан ниже:") for n, r in rows.items(): vals = [slope_jitter(r, w) for w in WINDOWS] print(f" {n:9} " + " ".join(f"{w} дн: {v:.2f} п.п." for w, v in zip(WINDOWS, vals))) print("\n4 · ТРИ ОКНА НА ОДНОМ И ТОМ ЖЕ ДНЕ") print(" Наклон в процентах цены за 20 дней и полуширина полосы:") for n, r in rows.items(): sp = window_spread(r, len(r)) cells = " ".join(f"{w} дн: {sp[w]['slope']:+6.1f}% ±{sp[w]['half']:4.1f}%" for w in WINDOWS) print(f" {n:9} {cells}") ups = 0 for n, r in rows.items(): sp = window_spread(r, len(r)) signs = {1 if sp[w]["slope"] > 0 else -1 for w in WINDOWS} if len(signs) > 1: ups += 1 print(f" Инструментов, где три окна дали РАЗНОЕ НАПРАВЛЕНИЕ канала: " f"{ups} из {len(rows)}") print("\nВЫВОД. Канал — это измерение с тремя настройками: длина окна,") print("ширина полосы и день, в который вы смотрите. Меняется любая —") print("меняется картинка, а вместе с ней и «очевидный» вывод. Полоса") print("в два отклонения охватывает почти всё прошлое, поэтому сама по") print("себе не выделяет ничего; касание границы возвращает цену") print("к середине примерно с той же частотой, что и любой другой день.") print("Польза канала не в предсказании, а в единице измерения: он даёт") print("число для «далеко» и «близко», одинаковое у всех, кто считает.") return 0 def _run() -> int: """Прогон со снимком: что напечатано, то и легло в `results.txt`. Пока файл клали руками, он отставал от скрипта, и сценарий писался по устаревшим числам. Здесь отставать нечему. Снимок пишется только при удачном прогоне. Обрубок опаснее отсутствия: он выглядит как настоящий отчёт. """ import io 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())