#!/usr/bin/env python3 """Все числа TW-025 одним прогоном. Тема L-08: разметка графика с нуля. ЗАЧЕМ ЭТОТ ФАЙЛ ЛЕЖИТ В ПРОЕКТЕ. Канал обещает: всё измерено, а не заявлено. Обещание стоит ровно столько, сколько стоит возможность пересчитать. ЧТО ЗДЕСЬ ПРОИСХОДИТ. Это финал модуля уровней, и он не вводит ни одного нового правила. Он берёт то, что уже измерено в трёх предыдущих роликах, и применяет к пустому графику — по шагам, до готового набора линий. TW-022: основание даёт кандидата. Излом сильнее края широкого бара (+6,7 против +0,4 пункта над сдвинутой отметкой). TW-023: непроверенный уровень лучше проверенного (−13,7 пункта за третье касание), созданный широким баром хуже обычного (−2,9), круглое число и возраст не значат ничего. TW-024: видимый уровень — место чистого пробоя, а не отскока. ОТБОР, КОТОРЫЙ ИЗ ЭТОГО СЛЕДУЕТ. Никаких весов, подобранных задним числом: каждый пункт взят из измерения выше и стоит ровно столько, сколько там намерено. Уровень получает балл за излом и за свежесть и теряет его за широкий бар и за то, что он шестидесятидневный экстремум. ЧЕСТНАЯ ПРОВЕРКА. Разметка делается на дату, у которой есть будущее: берётся точка за CHECK баров до конца ряда, размечается только по прошлому, а потом смотрится, что случилось дальше. Набор сравнивается с наивной разметкой — теми самыми заметными уровнями, которые размечают все. ДАННЫЕ. `04_assets/market/*-1d.json`. Разметка показывается на AAPL, проверка повторяется на всех шести инструментах. """ 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"] SHOW = "AAPL" # инструмент, который разбирается вслух в ролике CHECK = 120 # сколько дней будущего оставляем на проверку LOOK = 250 # сколько прошлого смотрим при разметке WING = 2 WIDE = 1.8 NEAR = 0.25 ROLL = 60 # обычный ход: медиана high−low за 60 баров до бара AWAY = 0.5 AHEAD = 5 SEEN = 60 TOP = 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 adr(r: list[dict]) -> list[float]: """Обычный дневной ход НА КАЖДЫЙ БАР: медиана high−low за ROLL баров ДО него. Одна медиана в долларах за весь ряд мерила годы, а не дни: цена Apple выросла в пять с половиной раз, и «широких» баров выходило 0 % в 2019-м и 60 % в 2026-м, а в разметку TW-025 попадало спрятанное будущее. Первые 20 баров ряда берут медиану первых двадцати. """ return [st.median([b["h"] - b["l"] for b in # без будущего (B-47): первые 20 баров — по тому, что было до них (r[max(0, i - ROLL):i] if i >= 20 else (r[:i] or r[:1]))]) for i in range(len(r))] def candidates(r: list[dict], upto: int, a: float, alive_only: bool = True) -> list[dict]: """Все основания в окне LOOK до точки разметки, с признаками. alive_only=False оставляет и пробитые: так считается ПОТОК появления кандидатов. Каждый излом в момент появления жив, поэтому скорость появления — это все изломы и края, а не доживших до разметки: живые, делённые на окно, недосчитывают старые основания, у которых было больше времени умереть. """ out = [] lo = max(WING, upto - LOOK) for i in range(lo, upto - WING): left, right = r[i - WING:i], r[i + 1:i + 1 + WING] wide = (r[i]["h"] - r[i]["l"]) > WIDE * a[i] for kind, price in (("high", r[i]["h"]), ("low", r[i]["l"])): turn = (all(price > b["h"] for b in left + right) if kind == "high" else all(price < b["l"] for b in left + right)) if not (turn or wide): continue w = r[max(0, i - SEEN):i + 1] seen = (price >= max(b["h"] for b in w) if kind == "high" else price <= min(b["l"] for b in w)) tests = 0 away = True dead = False # ИЗЛОМ ИЗВЕСТЕН ТОЛЬКО ЧЕРЕЗ WING БАРОВ (9 октября 2026). Его # определяют соседи справа, и до закрытия бара i + WING уровня ещё нет. # Поиск касания с бара i + 1 засчитывал как раз эти бары: у максимума они # по определению ниже его, часто в пределах допуска, и цена после них уже # уходит прочь — отскок им дарило само определение излома. В расчёте # TW-023 так легли 828 из 1701 первых касаний (49 %), и «отскакивали» они # в 97 % случаев против 81 % у остальных. Торговать их нельзя. # Край широкого бара известен на закрытии самого бара — ему ждать нечего. for j in range(i + 1 + (WING if turn else 0), upto): # УРОВЕНЬ ЖИВЁТ, ПОКА НЕ ПРОБИТ (A-40 второго аудита) — то же # правило, что в расчёте TW-023: закрытие дальше дневного хода # за уровнем делает его другим уровнем. Без этого правила # разметка держала 111 кандидатов, из которых живы были 35, и # все пять «лучших» линий умерли за месяцы до разметки. if (r[j]["c"] > price + a[j] if kind == "high" else r[j]["c"] < price - a[j]): dead = True break touching = r[j]["l"] - NEAR * a[j] <= price <= r[j]["h"] + NEAR * a[j] if touching and away: tests += 1 away = False elif not touching and abs(r[j]["c"] - price) > AWAY * a[j]: away = True if dead and alive_only: continue out.append({"i": i, "price": price, "kind": kind, "turn": turn, "wide": wide, "seen": seen, "tests": tests, "age": upto - i, "far": abs(r[upto - 1]["c"] - price) / a[upto]}) return out SPACING = 0.5 # две линии ближе полхода — одна зона (TW-022, раздел 2c) def spaced(ranked: list[dict], day: float) -> list[dict]: """Первые TOP линий, не ближе SPACING обычного дня друг к другу. ЗОНА, А НЕ ЦЕНТ (9 октября 2026). По расчёту TW-022 после правки окна подтверждения излома отметка, сдвинутая на полдня, отскакивает так же, как сам излом. Две линии ближе полхода — одна зона, и пятёрка, где их две, считает один уровень дважды (находка A-37 второго аудита: 224.69 и 223.78 стояли в 0,91 $ друг от друга). """ out: list[dict] = [] for c in ranked: if all(abs(c["price"] - p["price"]) >= SPACING * day for p in out): out.append(c) if len(out) == TOP: break return out def score(c: dict) -> float: """Балл уровня. Каждое слагаемое — из измерения предыдущих роликов. Величины взяты прямо оттуда, в процентных пунктах: столько эта черта прибавляла или отнимала к доле отскока. Ни один коэффициент не подобран здесь — иначе это была бы подгонка под тот же ряд, на котором проверяется. """ # ИЗ ПЯТИ СЛАГАЕМЫХ ОСТАЛОСЬ ОДНО (9 октября 2026). Прежние веса — # излом +7.7, широкий бар −2.9, проверки −13.7 / −10.7, видимость −3.0 — # были медианами расчётов TW-022 и TW-023, а те засчитывали касания # на двух барах после излома, когда излом ещё не подтверждён. После # правки излом против сдвинутого двойника, излом на широком баре и # видимость остались в пределах двух погрешностей — вес у них был бы # шумом. Померено одно: уровень, который цена уже проверяла хотя бы # раз, отскакивает реже непроверенного на 5.8 пункта (TW-023, раздел 7, # 3.4 погрешности, после правки B-47). Ничьи решает расстояние. return -5.8 if c["tests"] >= 1 else 0.0 def outcome(r: list[dict], c: dict, upto: int, a: float) -> str | None: """Что случилось с уровнем в оставленном будущем.""" near = NEAR * a[upto] # ход на дату разметки, без будущего for j in range(upto, min(upto + CHECK, len(r))): if r[j]["l"] - near <= c["price"] <= r[j]["h"] + near: seg = r[j + 1:min(j + 1 + AHEAD, len(r))] if not seg: return None if c["kind"] == "high": back = min(b["l"] for b in seg) <= c["price"] - AWAY * a[upto] else: back = max(b["h"] for b in seg) >= c["price"] + AWAY * a[upto] return "отскок" if back else "прошли насквозь" return "не дошли" 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']}") r = rows[SHOW] a = adr(r) upto = len(r) - CHECK print(f"\nРАЗМЕТКА {SHOW} НА {r[upto - 1]['d'][:10]}") print(f" Цена закрытия в этот день: {r[upto - 1]['c']:.2f}") print(f" Обычный дневной ход: {a[upto]:.2f} ({a[upto] / r[upto - 1]['c'] * 100:.2f}%)") print(f" Смотрим назад {LOOK} баров, вперёд не смотрим вовсе.") cand = candidates(r, upto, a) print(f"\n1 · СКОЛЬКО ОСНОВАНИЙ НАШЛОСЬ") print(f" всего: {len(cand)}") print(f" изломов: {sum(1 for c in cand if c['turn'])}, " f"краёв широких баров: {sum(1 for c in cand if not c['turn'])}") print(f" из них ни разу не проверены: " f"{sum(1 for c in cand if c['tests'] == 0)}") print(" Разметить всё — значит не разметить ничего.") near = [c for c in cand if c["far"] <= 8] print(f"\n2 · ОТСЕКАЕМ ДАЛЁКОЕ") print(" Уровень дальше восьми дневных ходов от цены — не решение") print(" на завтра, а украшение: цена дойдёт до него не скоро.") print(f" осталось: {len(near)} из {len(cand)}") ranked = sorted(near, key=lambda c: (-score(c), c["far"])) print(f"\n3 · ПЯТЬ ЛИНИЙ, КОТОРЫЕ ОСТАЮТСЯ НА ГРАФИКЕ") # ДАТА РОЖДЕНИЯ ЛИНИИ. «Не проверен» в этом уроке значит «не # проверен С ТЕХ ПОР, КАК ЛИНИЯ ПОЯВИЛАСЬ», а не «цена здесь не # бывала»: на тех же ценах свечи стояли и раньше. Раз ролик это # говорит, расчёт обязан напечатать, с какого дня идёт отсчёт — # иначе подпись в кадре сверить не с чем (находка A-05). print(f" {'цена':>9}{'балл':>7}{'что это':>26}{'до цены':>10}" f"{'дальше':>18}{'линия с':>14}") picked = spaced(ranked, a[upto]) for c in picked: what = ("излом" if c["turn"] else "край широкого бара") if c["wide"] and c["turn"]: what = "излом на широком баре" if c["seen"]: what += ", виден всем" print(f" {c['price']:9.2f}{score(c):7.1f}{what:>26}" f"{c['far']:9.1f}д{str(outcome(r, c, upto, a)):>18}" f"{r[c['i']]['d'][:10]:>14}") print(f"\n4 · ТО ЖЕ МЕСТО, НАИВНАЯ РАЗМЕТКА") print(" Пять самых заметных линий: шестидесятидневные экстремумы") print(" и уровни, которые уже проверялись — то, что бросается в глаза.") naive = sorted([c for c in near if c["seen"] or c["tests"] >= 1], key=lambda c: (-c["tests"], c["far"])) naive = spaced(naive, a[upto]) for c in naive: what = "виден всем" if c["seen"] else f"проверен {c['tests']}×" print(f" {c['price']:9.2f}{'':7}{what:>26}" f"{c['far']:9.1f}д{str(outcome(r, c, upto, a)):>18}") print(f"\n5 · ТА ЖЕ ПАРА РАЗМЕТОК НА ВСЕХ ШЕСТИ") print(f" Доля линий, давших отскок при первом же касании,") print(f" в следующие {CHECK} дней:") print(f" {'':11}{'балл: дошли':>12}{'отскок':>10}" f"{'наивно: дошли':>19}{'отскок':>10}" f"{'вблизи: дошли':>20}{'отскок':>10}") diffs = [] # ОБЩИЕ ДОЛИ ПО ВСЕМ ЛИНИЯМ (A-18 второго аудита). Медиана шести долей # по пять линий прыгает ступенями по двадцать процентов: «дошли с # восьмидесяти до ста» было медианой, а по всем линиям сдвиг меньше. pool = {"top": [], "nv": [], "mix": []} reach: dict[bool, list[bool]] = {False: [], True: []} for n, rr in rows.items(): aa = adr(rr) up = len(rr) - CHECK cc = [c for c in candidates(rr, up, aa) if c["far"] <= 8] if not cc: continue top = spaced(sorted(cc, key=lambda c: (-score(c), c["far"])), aa[up]) nv = sorted([c for c in cc if c["seen"] or c["tests"] >= 1], key=lambda c: (-c["tests"], c["far"])) nv = spaced(nv, aa[up]) # ТРЕТИЙ ВАРИАНТ — тот же балл, но только среди достижимого. # Не подгонка весов, а исправление постановки: сначала отбираем # линии, до которых цена вообще может дойти, и только потом # ранжируем их качеством. rc = [c for c in cc if c["far"] <= 3] mix = spaced(sorted(rc, key=lambda c: (-score(c), c["far"])), aa[up]) def hit(group): res = [outcome(rr, c, up, aa) for c in group] got = [x for x in res if x in ("отскок", "прошли насквозь")] reach = len(got) / len(res) * 100 if res else 0.0 bounce = (sum(1 for x in got if x == "отскок") / len(got) * 100 if got else 0.0) return reach, bounce (r1, h1), (r2, h2), (r3, h3) = hit(top), hit(nv), hit(mix) for key, group in (("top", top), ("nv", nv), ("mix", mix)): pool[key] += [outcome(rr, c, up, aa) for c in group] # МЕХАНИЗМ УРОКА МЕРЯЕТСЯ НА ВСЕХ КАНДИДАТАХ, а не на тридцати # линиях: дошла ли цена до непроверенной линии реже, чем до # проверенной (9 октября 2026). for c in cc: reach[c["tests"] >= 1].append( outcome(rr, c, up, aa) in ("отскок", "прошли насквозь")) diffs.append((r1, h1, r2, h2, r3, h3)) print(f" {n:9} {r1:9.0f}% {h1:9.0f}% {r2:9.0f}% {h2:9.0f}%" f" {r3:11.0f}% {h3:9.0f}%") if diffs: print(f" → дошли: по баллу {st.median([d[0] for d in diffs]):.0f}%, " f"наивно {st.median([d[2] for d in diffs]):.0f}%") print(f" → отскок среди дошедших: по баллу " f"{st.median([d[1] for d in diffs]):.0f}%, " f"наивно {st.median([d[3] for d in diffs]):.0f}%") print(f" → тот же балл, но только вблизи (до 3 дневных ходов): " f"дошли {st.median([d[4] for d in diffs]):.0f}%, " f"отскок {st.median([d[5] for d in diffs]):.0f}%") print(" (выше — медианы шести долей; ниже — по всем линиям сразу)") for key, label in (("top", "по баллу"), ("nv", "наивно"), ("mix", "вблизи")): res = pool[key] got = [x for x in res if x in ("отскок", "прошли насквозь")] bnc = sum(1 for x in got if x == "отскок") print(f" {label:9} линий {len(res):3}: дошли {len(got)} " f"({len(got) / len(res) * 100:.0f}%), отскок {bnc} " f"({bnc / len(got) * 100 if got else 0:.0f}% дошедших)") u_, t_ = reach[False], reach[True] reach_untested = sum(u_) / len(u_) * 100 reach_tested = sum(t_) / len(t_) * 100 pu, pt = (sum(u_) + 0.5) / (len(u_) + 1), (sum(t_) + 0.5) / (len(t_) + 1) reach_err = ((pu * (1 - pu) / len(u_) + pt * (1 - pt) / len(t_)) * 1e4) ** 0.5 print(f" все кандидаты до 8 дневных ходов: до непроверенных цена дошла в" f" {reach_untested:.0f}% из {len(u_)}, до проверенных — в" f" {reach_tested:.0f}% из {len(t_)}" f" ({reach_untested - reach_tested:+.0f} ± {reach_err:.0f} п.п.)") print(" ВНИМАНИЕ: пять линий на инструмент — это единицы событий.") print(" Здесь это иллюстрация метода, а не доказательство: померенное") print(" в трёх предыдущих роликах стоит на тысячах касаний.") print("\n6 · КАК БЫСТРО РАЗМЕТКА УСТАРЕВАЕТ") print(" Каждый новый бар добавляет основания и двигает цену, так что") print(" набор линий — не разовая работа. Скорость появления:") for n, rr in rows.items(): aa = adr(rr) up = len(rr) - CHECK cc = candidates(rr, up, aa) born = candidates(rr, up, aa, alive_only=False) per_day = len(born) / LOOK live = [c for c in cc if c["far"] <= 3] print(f" {n:9} {per_day:.1f} новых кандидата в день " f"({per_day * (7 if n.endswith('USDT') else 5):.0f} за неделю), в живой зоне сейчас " f"{len(live)} (живых всего {len(cc)} из {len(born)})") print(" Живая зона — три дневных хода от цены. Всё остальное ждёт.") print("\nВЫВОД, ЗАПИСАННЫЙ ПОСЛЕ ПРОГОНА, И ОН НЕ РЕКЛАМНЫЙ.") print("") # ЧИСЛА ВЫВОДА ВЫЧИСЛЯЮТСЯ, А НЕ ПИШУТСЯ СЛОВАМИ. Прежний текст # держал «восемьдесят против ста» и «с семидесяти одного до семидесяти # трёх» — после правила «жив, пока не пробит» (A-40) они устарели, а # вывод отчитывался ими дальше. 9 октября 2026 то же случилось со # словами: «доходит РЕЖЕ» стояло при равных долях — теперь и сравнение # считается. def share(key): res = pool[key] got = [x for x in res if x in ("отскок", "прошли насквозь")] bnc = sum(1 for x in got if x == "отскок") return (len(got) * 100 / len(res) if res else 0.0, bnc * 100 / len(got) if got else 0.0) def cmp(a_, b_): return ("реже, чем" if a_ < b_ - 0.5 else "чаще, чем" if a_ > b_ + 0.5 else "так же часто, как") (rt, bt), (rn, bn), (rm, bm) = share("top"), share("nv"), share("mix") print(f"По баллу ({len(pool['top'])} линий): цена дошла до {rt:.0f}%," f" отскок у {bt:.0f}% дошедших.") print(f"Наивно ({len(pool['nv'])} линий): дошла до {rn:.0f}%," f" отскок у {bn:.0f}% дошедших.") print(f"Только вблизи, до трёх дневных ходов ({len(pool['mix'])} линий):" f" дошла до {rm:.0f}%, отскок у {bm:.0f}%.") print(f"До линий по баллу цена доходит {cmp(rt, rn)} до наивных;" f" вблизи — {cmp(rm, rt)} по баллу.") print("") print("ДВА РАЗНЫХ ВОПРОСА. Первый: дойдёт ли цена до линии вообще.") print("Второй: что будет, если дойдёт. Всё, что измерялось в модуле,") print("отвечало на второй.") d_ = reach_untested - reach_tested print(f"Единственный померенный признак — «ещё не проверяли» — на всех" f" кандидатах шести инструментов: до непроверенных цена дошла в" f" {reach_untested:.0f}% случаев, до проверенных — в {reach_tested:.0f}%" f" ({d_:+.0f} ± {reach_err:.0f} п.п., " f"{'померено' if abs(d_) >= 2 * reach_err else 'не померено'}).") print("") print("ЧЕСТНО ПРО САМО СРАВНЕНИЕ. Пять линий на инструмент — это тридцать") 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())