#!/usr/bin/env python3 """Все числа TW-026 одним прогоном. Тема K-06: один сетап на двух рынках. ЗАЧЕМ ЭТОТ ФАЙЛ ЛЕЖИТ В ПРОЕКТЕ. Канал обещает: всё измерено, а не заявлено. Обещание стоит ровно столько, сколько стоит возможность пересчитать. ЧТО ПРОВЕРЯЕТСЯ. Берётся ОДИН сетап, описанный полностью и одинаково для обоих рынков, и прогоняется на трёх акциях и трёх монетах. Вопрос не «где он работает лучше», а что именно в его арифметике меняется при переносе. СЕТАП. Тот, который построен предыдущими модулями, без единого нового правила: · вход — возврат к непроверенному уровню (излом, ни разу не тронутый с момента появления, в пределах трёх дневных ходов от цены); · стоп — за уровнем на буфер в половину обычного дневного хода; · цель — два стопа; · выход по времени — десять дней. ЧТО МЕРЯЕТСЯ И ПОЧЕМУ ИМЕННО ЭТО. 1. Доля выигрышных сделок и результат в R. Это то, что все и сравнивают. 2. Частота сделок: сколько сигналов в год. Стратегия, дающая три сделки в год, и стратегия, дающая тридцать, — разные занятия, даже при одинаковом проценте. 3. Размер стопа в процентах цены — то, что меняет размер позиции. 4. Издержки в долях R: комиссия и проскальзывание съедают тем больше, чем ближе стоп. Здесь два рынка расходятся сильнее всего. 5. Один и тот же результат в R после вычета издержек каждого рынка. ЧЕГО ЗДЕСЬ НЕТ. Плеча (это TW-020), фандинга и налогов: они зависят от площадки, а не от рынка. ДАННЫЕ. `04_assets/market/*-1d.json`, шесть инструментов, общее окно — с начала крипто-рядов, чтобы сравнивать одни и те же годы (урок TW-024). """ 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" STOCKS = ["AAPL", "TSLA", "SPY"] CRYPTO = ["BTCUSDT", "ETHUSDT", "SOLUSDT"] WING = 2 NEAR = 0.25 # подход к уровню ROLL = 60 # обычный ход: медиана high−low за 60 баров до бара BUFFER = 0.5 # буфер стопа за уровнем, в дневных ходах TARGET = 2.0 # цель в стопах HOLD = 10 # выход по времени LIVE = 3.0 # «живая зона»: сколько дневных ходов до уровня FEE_STOCK = 0.0005 # комиссия+проскальзывание на круг, доля позиции FEE_CRYPTO = 0.0020 BUFFERS = [0.5, 1.0, 1.5, 2.0] # перебор зазора между уровнем и стопом 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 trades(r: list[dict], a: float, buffer: float = BUFFER, flaw: str | set[str] | None = None) -> list[dict]: """Сделки по сетапу. Одно определение на оба рынка, без исключений. `flaw` воспроизводит ошибку одной из прежних версий расчёта — те, что давали плюс. Нужен затем, чтобы фраза «три версии врали» в ролике опиралась на прогон, а не на память: · "no_entry_day" — стоп в день входа не проверялся вовсе; · "entry_day_target" — в день входа засчитывалась и цель; · "loose_fill" — вход с допуском в четверть дневного хода и без проверки стороны подхода. ПРИНИМАЕТ И НАБОР. Ролик рассказывает не три отдельные ошибки, а ЦЕПОЧКУ: сперва были все, потом их убирали по одной. Стадия — это набор, и каждая ошибка обязана действовать сама по себе. Пока флаг был один, `first` гасился любым из них, и «только слабый вход» посчитать было нельзя: в нём всегда засчитывалась и цель в день входа. """ flaws = ({flaw} if isinstance(flaw, str) else set(flaw) if flaw else set()) out = [] for i in range(WING, len(r) - WING - HOLD - 1): left, right = r[i - WING:i], r[i + 1:i + 1 + WING] for kind, price in (("high", r[i]["h"]), ("low", r[i]["l"])): ok = (all(price > b["h"] for b in left + right) if kind == "high" else all(price < b["l"] for b in left + right)) if not ok: continue # Первое касание — и оно же вход. Уровень должен быть свежим: # если цена уже приходила, сетапа нет (измерено в TW-023). hit = None for j in range(i + 1, min(i + 1 + 120, len(r) - HOLD - 1)): # ВХОД ТОЛЬКО ЕСЛИ ЦЕНА ДЕЙСТВИТЕЛЬНО ДОШЛА. С допуском # в четверть дневного хода заявка «исполнялась» там, куда # цена не добралась, — и вход получал цену лучше доступной. slack = NEAR * a[j] if "loose_fill" in flaws else 0.0 if r[j]["l"] - slack <= price <= r[j]["h"] + slack: hit = j break if hit is None: continue if abs(r[hit - 1]["c"] - price) / a[hit] > LIVE: continue # уровень был далеко — не сетап # СТОРОНА ПОДХОДА. К сопротивлению цена должна прийти снизу, # к поддержке — сверху. Без этой проверки продажа «от уровня» # засчитывалась и тогда, когда цена давно торговалась выше него, # то есть когда уровня как сопротивления уже не существовало. prev = r[hit - 1]["c"] if "loose_fill" not in flaws: if kind == "high" and prev >= price: continue if kind == "low" and prev <= price: continue stop = (price + buffer * a[hit] if kind == "high" else price - buffer * a[hit]) risk = abs(price - stop) # ВХОД, УЖЕ СТОЯЩИЙ ЗА СТОПОМ, здесь невозможен по построению: # вход идёт ПО цене уровня лимитной заявкой, а стоп отстоит # от неё на buffer дневных ходов. Проверка оставлена явной, # чтобы при смене входа на «по открытию» её не забыли. if risk <= 0: continue target = (price - TARGET * risk if kind == "high" else price + TARGET * risk) res = None # ДЕНЬ ВХОДА СЧИТАЕТСЯ АСИММЕТРИЧНО, и это не придирка. # Первая версия начинала со следующего дня: сделки, где стоп # выносило сразу, молча доживали до завтра. Вторая считала # в день входа и стоп, и цель — и доля выигрышных прыгнула # до семидесяти пяти процентов, потому что цена, дошедшая # до цели в тот же день, скорее всего была там ДО касания # уровня, а не после. # Порядок событий внутри дня четыре числа бара не содержат # (это весь TW-018), поэтому берётся худший вариант: в день # входа засчитывается только стоп, цель — со следующего дня. begin = hit + 1 if "no_entry_day" in flaws else hit for j in range(begin, min(hit + 1 + HOLD, len(r))): first = (j == hit and "entry_day_target" not in flaws) if kind == "high": if r[j]["h"] >= stop: res = -1.0 break if not first and r[j]["l"] <= target: res = TARGET break else: if r[j]["l"] <= stop: res = -1.0 break if not first and r[j]["h"] >= target: res = TARGET break if res is None: # вышли по времени last = r[min(hit + HOLD, len(r) - 1)]["c"] res = ((price - last) / risk if kind == "high" else (last - price) / risk) out.append({"i": hit, "res": res, "risk": risk, "kind": kind, "price": price, "stop_pct": risk / price * 100}) # ОДИН ДЕНЬ — ОДНА СДЕЛКА В СТОРОНУ. Один бар может задеть три излома # подряд, и каждый запишется отдельной сделкой с почти одинаковым # исходом: выборка выглядит независимее, чем есть. Из совпавших # остаётся уровень, ближайший к закрытию дня. # Найдено 10 сентября 2026 на расчёте TW-030 и проверено здесь. best = {} for g in out: key = (g["i"], g["kind"]) # ВЧЕРАШНЕЕ ЗАКРЫТИЕ, А НЕ ЗАКРЫТИЕ ДНЯ КАСАНИЯ (B-47, 9 октября 2026). # При входе по касанию закрытия этого дня ещё нет, а выбор по нему # сдвигал результат: у TW-027 вход по касанию со стопом в ход и трейлом # −0.028R → −0.063R, у TW-034 вилка 0.144R → 0.203R. Выбор по открытию # и по вчерашнему закрытию дают одно и то же — выбросом было закрытие. near = abs(r[g["i"] - 1]["c"] - g["price"]) if key not in best or near < best[key]["_near"]: best[key] = dict(g, _near=near) return [{k: v for k, v in g.items() if k != "_near"} for g in sorted(best.values(), key=lambda x: x["i"])] 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 STOCKS + CRYPTO} start = min(r[0]["d"][:10] for n, r in rows.items() if n in CRYPTO) rows = {n: [b for b in r if b["d"][:10] >= start] for n, r in rows.items()} print("ДАННЫЕ") print(f" Общее окно с {start} — одни и те же годы для обоих рынков.") for n, r in rows.items(): print(f" {n:9} {len(r)} баров {r[0]['d']} … {r[-1]['d']}") data = {n: trades(r, adr(r)) for n, r in rows.items()} print("\n1 · ОДИН И ТОТ ЖЕ СЕТАП, ШЕСТЬ ИНСТРУМЕНТОВ") print(f" {'':11}{'сделок':>8}{'в год':>8}{'выигрышных':>13}" f"{'итог, R':>10}{'на сделку':>12}") for n, r in rows.items(): t = data[n] if not t: continue years = len(r) / (365 if n in CRYPTO else 252) wins = sum(1 for x in t if x["res"] > 0) / len(t) * 100 total = sum(x["res"] for x in t) print(f" {n:9} {len(t):7} {len(t) / years:7.0f} {wins:12.0f}%" f" {total:9.1f} {total / len(t):11.2f}") print("\n2 · ТОТ ЖЕ СЕТАП ПРИ РАЗНОМ БУФЕРЕ СТОПА") print(" Буфер — расстояние от уровня до стопа, в дневных ходах.") print(" Это единственная настройка, которая здесь меняется.") print(f" {'':11}" + "".join(f"{str(b) + ' ATR':>12}" for b in BUFFERS)) # ИЗДЕРЖКИ — ДЛЯ КАЖДОГО БУФЕРА СВОИ (A-41 второго аудита). В долях R # их задаёт ширина стопа, а она растёт с буфером; голос s27 говорил # «costs eat nearly all of it» по издержкам базового буфера. buf: dict[str, dict[float, tuple[float, float]]] = {} for n, r in rows.items(): aa = adr(r) cells = [] buf[n] = {} fee = FEE_CRYPTO if n in CRYPTO else FEE_STOCK for b in BUFFERS: t = trades(r, aa, b) avg = sum(x["res"] for x in t) / len(t) if t else 0.0 c = fee / (st.median([x["stop_pct"] for x in t]) / 100) if t else 0.0 buf[n][b] = (avg, c) cells.append(avg) print(f" {n:9} " + "".join(f"{c:+11.2f}R" for c in cells)) print(" (средний результат сделки в R, до издержек)") print(" после издержек своего буфера:") for n in rows: print(f" {n:9} " + "".join(f"{buf[n][b][0] - buf[n][b][1]:+11.3f}R" for b in BUFFERS)) above = sum(1 for n in rows for b in BUFFERS if buf[n][b][0] > 0) above_net = sum(1 for n in rows for b in BUFFERS if buf[n][b][0] - buf[n][b][1] > 0) best_net = max(buf[n][b][0] - buf[n][b][1] for n in rows for b in BUFFERS) print(f" выше нуля до издержек {above} из {len(rows) * len(BUFFERS)}, " f"после — {above_net}; лучшая клетка после издержек {best_net:+.3f}R") print("\n3 · ЧТО НЕ МЕНЯЕТСЯ") sw = [sum(1 for x in data[n] if x["res"] > 0) / len(data[n]) * 100 for n in STOCKS if data[n]] cw = [sum(1 for x in data[n] if x["res"] > 0) / len(data[n]) * 100 for n in CRYPTO if data[n]] print(f" доля выигрышных: акции {min(sw):.0f}–{max(sw):.0f}%, " f"крипта {min(cw):.0f}–{max(cw):.0f}%") se = [sum(x["res"] for x in data[n]) / len(data[n]) for n in STOCKS if data[n]] ce = [sum(x["res"] for x in data[n]) / len(data[n]) for n in CRYPTO if data[n]] print(f" R на сделку: акции {min(se):+.2f}…{max(se):+.2f}, " f"крипта {min(ce):+.2f}…{max(ce):+.2f}") print("\n4 · ЧТО МЕНЯЕТСЯ: РАЗМЕР СТОПА В ПРОЦЕНТАХ ЦЕНЫ") print(" Именно он задаёт размер позиции при одном и том же риске.") for n in STOCKS + CRYPTO: t = data[n] if not t: continue m = st.median([x["stop_pct"] for x in t]) print(f" {n:9} {m:5.2f}% при риске 1% счёта позиция = " f"{1 / m * 100:5.0f}% счёта") print("\n5 · ЧТО МЕНЯЕТСЯ СИЛЬНЕЕ ВСЕГО: ИЗДЕРЖКИ В ДОЛЯХ R") print(f" Круг стоит {FEE_STOCK:.2%} позиции на акциях и " f"{FEE_CRYPTO:.2%} на крипте.") print(" В долях R это зависит от того, насколько близко стоп:") print(f" {'':11}{'стоп, %':>9}{'издержки, R':>14}{'на 30 сделок':>15}") cost = {} for n in STOCKS + CRYPTO: t = data[n] if not t: continue m = st.median([x["stop_pct"] for x in t]) / 100 fee = FEE_CRYPTO if n in CRYPTO else FEE_STOCK c = fee / m cost[n] = c print(f" {n:9} {m * 100:8.2f}% {c:13.3f}R {c * 30:14.1f}R") print("\n6 · ТОТ ЖЕ РЕЗУЛЬТАТ ПОСЛЕ ИЗДЕРЖЕК") print(f" {'':11}{'до издержек':>14}{'после':>10}{'съедено':>10}") for n in STOCKS + CRYPTO: t = data[n] if not t or n not in cost: continue raw = sum(x["res"] for x in t) / len(t) net = raw - cost[n] print(f" {n:9} {raw:+13.2f}R {net:+9.2f}R {cost[n]:9.3f}R") # ВЫВОД СЧИТАЕТСЯ, А НЕ ПОМНИТСЯ. Прежде здесь стояли те же # утверждения числами, набранными словами. Таблицы выше пересчитываются, # а слова — нет: правило в расчёте поменяли, таблицы уехали, проза # осталась, и ролик разошёлся с собственным расчётом. Теперь каждая # величина берётся из того же прогона, что и таблица над ней. per = {n: sum(x["res"] for x in data[n]) / len(data[n]) for n in STOCKS + CRYPTO if data[n]} grid = [] for n, r in rows.items(): aa = adr(r) for b in BUFFERS: tr = trades(r, aa, b) if tr: grid.append(sum(x["res"] for x in tr) / len(tr)) freq = {n: len(data[n]) / (len(rows[n]) / (365 if n in CRYPTO else 252)) for n in STOCKS + CRYPTO if data[n]} med = {n: st.median([x["stop_pct"] for x in data[n]]) for n in STOCKS + CRYPTO if data[n]} thin = min(med, key=med.get) wide = max(med, key=med.get) # ЦЕПОЧКА ОТЛАДКИ — прогоном, а не по памяти. Стадия это НАБОР # ошибок: сперва были все, потом их убирали по одной. Ровно та # последовательность, которую рассказывает ролик, поэтому здесь же # считается и доля выигрышных — её голос называет на каждой стадии. STAGES = ( ("первый прогон: день входа не смотрели вовсе", {"no_entry_day", "loose_fill"}), ("проверяем и стоп, и цель в день входа", {"entry_day_target", "loose_fill"}), ("день входа несимметричен: только стоп", {"loose_fill"}), ("честный расчёт", set()), ) chain = [] for label, flaws in STAGES: got, tot, win = [], 0, 0 for n, r in rows.items(): tr = trades(r, adr(r), BUFFER, flaw=flaws) if not tr: continue got.append(sum(x["res"] for x in tr) / len(tr)) tot += len(tr) win += sum(1 for x in tr if x["res"] > 0) chain.append((label, tot, win / tot * 100 if tot else 0.0, sum(got) / len(got) if got else 0.0)) # СВЕРКА. Последняя стадия — это и есть главный расчёт этюда, тот # самый, из которого собраны таблицы выше. Разошлись — значит флаг # протёк в честный путь, и печатать цепочку нельзя. honest = sum(per.values()) / len(per) if abs(chain[-1][3] - honest) > 0.005: raise SystemExit(f"ПЛОХО: честная стадия {chain[-1][3]:+.3f} R" f" против главного расчёта {honest:+.3f} R") print("\nВЫВОД, ПОСЧИТАННЫЙ ЭТИМ ЖЕ ПРОГОНОМ.") print("") print("ГЛАВНОЕ И НЕПРИЯТНОЕ. Сетап, собранный из всего, что канал") print("измерил в модуле уровней, на дневных барах УБЫТОЧЕН — и на") print(f"акциях, и на крипте: {sum(1 for c in grid if c < 0)} из {len(grid)}") print("сочетаний «инструмент × буфер» ниже нуля.") print(f"До издержек от {min(grid):+.2f} до {max(grid):+.2f} R на сделку;") print("после издержек хуже везде.") print("") print("ТРИ ВЕРСИИ ЭТОГО РАСЧЁТА ДАВАЛИ ПЛЮС, И ВСЕ ТРИ ВРАЛИ.") print("Ниже вся цепочка отладки: сперва ошибки были все, потом") print("убирались по одной. Числа не из памяти — этот же прогон,") print("среднее по шести инструментам. «Выигр.» — доля от ВСЕХ сделок") print("стадии, а не по инструменту, как в таблице 1.") print(f" {'стадия':44} {'сделок':>7} {'выигр.':>7} {'R/сделку':>9}") for label, tot, wr, avg in chain: print(f" {label:44} {tot:7} {wr:6.0f}% {avg:+9.2f}") print(" — цена, дошедшая до цели в день касания, скорее была там") print(" ДО касания, а не после; вход с допуском — это цена,") print(" до которой рынок не дошёл, и уровень, которого как") print(" уровня уже не было.") print(f" Выборка падает с {chain[0][1]} до {chain[-1][1]} на последней") print(" стадии: исчезают сделки, которых не было, — вход по цене,") print(" до которой рынок не дошёл.") print("") print("ЧТО НЕ МЕНЯЕТСЯ МЕЖДУ РЫНКАМИ.") print(f" доля выигрышных: акции {min(sw):.0f}–{max(sw):.0f} %," f" крипта {min(cw):.0f}–{max(cw):.0f} %") print(" знак результата: отрицательный на обоих рынках") # ОТКЛИК НА БУФЕР СЧИТАЕТСЯ (A-19). Здесь стояла вписанная строка # «шире зазор — чуть менее убыточно», а при обычном ходе по 60 барам # от половины хода к целому акции улучшаются, Bitcoin и Ethereum — нет. print(" отклик на буфер — сколько инструментов теряют меньше," f" чем при {BUFFERS[0]} хода:") for b in BUFFERS[1:]: st_ = sum(1 for n in STOCKS if buf[n][b][0] > buf[n][BUFFERS[0]][0]) cr_ = sum(1 for n in CRYPTO if buf[n][b][0] > buf[n][BUFFERS[0]][0]) print(f" при {b} хода: акции {st_} из {len(STOCKS)}," f" крипта {cr_} из {len(CRYPTO)}") print("") print("ЧТО МЕНЯЕТСЯ. Размер стопа в процентах цены — от") print(f"{med[thin]:.2f} % у {thin} до {med[wide]:.2f} % у {wide}: при одном") print("и том же риске в один процент счёта позиция отличается") print(f"в {med[wide] / med[thin]:.1f} раза.") print(f"Частота: на крипте {min(freq[n] for n in CRYPTO if n in freq):.0f}–" f"{max(freq[n] for n in CRYPTO if n in freq):.0f} сделки в год") print(f"против {min(freq[n] for n in STOCKS if n in freq):.0f}–" f"{max(freq[n] for n in STOCKS if n in freq):.0f} на акциях.") print("") # СРАВНИВАТЬ ИЗДЕРЖКИ НАДО ВНУТРИ РЫНКА. Через рынок комиссия разная # (0,05 % против 0,20 %), и разброс перестаёт быть про тесноту стопа: # у BTCUSDT стоп ШИРЕ, чем у TSLA, а платит он вчетверо больше — # просто потому, что круг на крипте дороже. Прежняя формулировка # «зависят не от рынка, а от тесноты стопа» на числах не держалась. print("И СИЛЬНЕЕ ВСЕГО — ИЗДЕРЖКИ. В долях R их задаёт теснота стопа,") print("и это видно внутри каждого рынка, где комиссия одна и та же:") for group, name in ((STOCKS, "акции"), (CRYPTO, "крипта")): g = [n for n in group if n in cost] hi, lo = max(g, key=lambda n: cost[n]), min(g, key=lambda n: cost[n]) print(f" {name}: {hi} платит {cost[hi]:.3f} R при стопе" f" {med[hi]:.2f} %, {lo} — {cost[lo]:.3f} R при" f" {med[lo]:.2f} % ({cost[hi] / cost[lo]:.1f}×)") print("Между рынками складываются обе причины: круг на крипте дороже") print(f"вчетверо, и потому {max(cost, key=cost.get)} платит больше всех —") print(f"{max(cost.values()):.3f} R за сделку, при стопе не самом тесном.") 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())