# Щоденник дослідження /wclass — як знайти класи станів ринку і навчити їх детектувати **Що це.** Виконуваний рецепт. Читаєш цей файл, береш свій датасет — отримуєш ті самі класи станів ринку і навчені моделі, що їх детектують. Жодного захардкодженого списку вікон, жодного «клас 1 — це ось ці». Клас визначається **процедурою**, і процедура перерахує його на будь-яких даних: іншій монеті, іншому періоді, більшій історії. Числа, що трапляються в тексті — це те, що вийшло **на нашому зрізі** (BTC, 171 доба). Вони наведені як орієнтир «схоже чи ні», а не як константи. Твій прогін дасть свої. --- ## 0. Вхід Посекундні дані ордербука, 8 фіч + ціна: ``` const_resist resist_plus resist_minus — стіна опору: товщина, приплив, відтік const_support support_plus support_minus — стіна підтримки: те саме vol_buy vol_sell — реальні угоди: маркет-баї і маркет-сели price_avg — ціна (у вхід моделі НЕ входить, лише перевірка) timestamp — секунда ``` Формат у нас: `{DATA}/2026/{MM}/{COIN}_USDT/1S_YYYY-MM-DD_COIN_USDT.csv`, один файл на добу. Мінімум для повторення — **≥60 діб** посекундних даних однієї монети. Менше — класи знайдуться, але перевірки на стійкість не матимуть сенсу. **Правило, яке визначає всю конструкцію:** вікно класифікується **тільки за своїми 1680 числами** (8 фіч × 210 секунд). Ніякої багатодобової норми, ціни, часу доби чи сусідніх вікон на вході моделі. Дозволені лише параметри, вивчені на історії — ваги мережі, дерева, осі, пороги; вони заморожені й у бою не перераховуються. --- ## 1. Крок перший: нарізати вікна `python3 wclass.py data COIN` Три речі, без яких далі все буде брехнею: 1. **Дедуплікація timestamp.** У сирих даних трапляються повтори (у нас 192 тис. рядків із 12.2 млн). Без цього «210 секунд» місцями означає менше 210 секунд. 2. **Суцільна 1-секундна шкала.** Частина секунд фізично відсутня (у нас 18.4%). Ряд кладеться на сітку від першої до останньої секунди, дірки заповнюються лінійно і позначаються маскою. Вікно береться, лише якщо **≥95% його секунд справжні**. 3. **Спліт блоками з ембарго.** Сусідні вікна корелюють ~0.76. Блоки по 2 години перемішуються, 10% у валідацію, і все, що ближче 30 хв до валідаційного вікна, викидається з навчання. Випадковий спліт по одиночних вікнах завищить усі оцінки. Вікна ріжуться **стик у стик**, крок = вікно, тож між ними не губиться жодна секунда. На виході: `X` розміру (вікон, 8, 210) — сирі значення; `ts`, `px`, індекси train/val, і `rng_fwd` — розмах ціни в наступну годину після вікна (це «конверт», відкривається лише після того, як класи знайдені). --- ## 2. Крок другий: описати вікно `python3 wclass.py feat COIN` Це **єдине місце, де є свобода**, і саме тут вирішується, що конвеєр здатен побачити. Сирий ряд фічі — невідʼємний, з нулями і важким хвостом; звичайне середнє по ньому ламає один тик. Тому вікно описується статистиками, про які відомо, що вони міряють. На кожну з 8 фіч: | опис | формула | що міряє | |---|---|---| | `lev` | середнє `log1p(x/s)` за вікно | рівень | | `sd` | розкид того ж | нерівність у часі | | `conc` | сума 10 найбільших секунд ÷ вся сума | концентрація: один сплеск чи рівний потік | | `quiet` | частка секунд нижче 5% масштабу | скільки часу фіча мовчить | | `trend` | 1-й ортонормований поліном Лежандра | нахил усередині вікна | | `curv` | 2-й поліном | вигин | | `ac1` | автокореляція лаг-1 | інерція | Плюс **склад вікна**: `clr` від 8 нормованих мас — центроване лог-відношення, правильна алгебра для часток (сума логарифмів мінус їхнє середнє). І `size` — загальний «розмір» вікна, та частина, яку `clr` відкидає. Разом 89 чисел. Ортонормовані поліноми будуються через QR-розклад, а не формулою — так гарантується точна ортогональність до константи. **Три подання** з цих 89 описів. Вони потрібні, щоб питання «клас — це склад ринку чи просто рівень активності?» закривалось числом: - `full` — усе; - `free` — без абсолютних рівнів (нормовані форма й склад); - `lev` — лише рівні. --- ## 3. Крок третій: чи є класи взагалі `python3 wclass.py model COIN` · `tests` · `raws` · `ladder` **Спершу треба зʼясувати, чи класи існують.** Якщо кластеризувати без цієї перевірки, алгоритм намалює межі в будь-якому випадку — навіть на чистому шумі. Кожен критерій рахується **разом зі своїм нуль-контролем**, прогнаним через увесь конвеєр: - **жорсткий нуль** — клітинки (фіча, секунда) перемішуються МІЖ вікнами. Гине все, крім маргінальних розподілів. Перемішувати окремо в TRAIN і у VAL, інакше витік. - **мʼякий нуль** — перемішуються цілі ряди фіч. Форма кожної фічі всередині вікна ціла, спільність фіч зруйнована. Різниця між двома нулями показує, **де саме** сидить структура. Критерії: 1. **Gap-статистика** (Тібширані) з нуль-контролем як еталоном замість рівномірного куба: `log W(нуль) − log W(дані)` для K = 1..12. Наш еталон чесніший — він зберігає маргіналії. 2. **Силует** на тих самих K, дані проти нуля. 3. **Тест Сільвермана** на мультимодальність: критична ширина ядра + згладжений бутстреп. Рахується на перших головних осях і на напрямку максимального розділення (напрямок Фішера між двома k-середніми), причому **на відкладених вікнах** — напрямок обирався на навчанні. 4. **Стійкість** — ARI між мітками двох **ПОВНИХ перебудов** конвеєра на непересічних часових половинах (вінзоризація, масштаб, PCA, кластеризація — заново), зіставлених на спільній валідації. Швидка оцінка з фіксованим простором завищує вдвічі. **Що ми отримали і що це означає:** gap ≈ 0 у просторах складу, силует на рівні шуму, Сільверман не знайшов другого горба в жодній осі, стійкість низька, а при K = 5-6 нуль-контроль СТІЙКІШИЙ за реальні дані. **Дискретних класів немає — це континуум.** При цьому структура є: 16 осей ловлять 83% дисперсії описів проти 62% у нуля; на сирому векторі 1680 — 31% проти 1.2%. Драбина нулів каже, де вона: реальні 0.831 → мʼякий нуль 0.627 → жорсткий 0.617. Тобто **структура в складі вікна, а часова форма додає ~1%**. ### Дві пастки, які тут неминуче зустрінуться - **Бутстреп-LRT не доводить наявність класів.** Він упевнено відкидає K=1 — і робить це однаково на реальних даних і на перемішаних клітинках. Він міряє негаусовість маргіналій. - **ICL так само.** Спадає монотонно до максимального K без внутрішнього оптимуму. Обидва критерії поважні й у літературі стандартні. Тут вони непридатні, і побачити це можна лише прогнавши їх по нуль-контролю. --- ## 4. Крок четвертий: як знайдено класи «шторм», «штиль», «клас 2» Оскільки класів як таких немає, розбиття — це **розріз континууму**. Він робиться детерміновано і відтворюється на будь-яких даних: 1. Простір: описи `full` → вінзоризація по квантилях TRAIN (0.1% / 99.9%) → робастний z (медіана й міжквартильний розмах TRAIN) → **PCA 16 з вибіленням**, підігнаний ТІЛЬКИ на TRAIN. Рангове перетворення навмисно не беремо: воно робить маргіналії гаусовими і стирає саме ту двогорбість, яку ми перевіряємо. 2. **k-середні, K = 4**, навчені на TRAIN, застосовані до всіх вікон. 3. **Іменування — з боку ВХОДУ, а не за номером кластера.** Номери k-середніх довільні й міняються між прогонами, тому клас визначається рангом за сумарним рівнем шести потокових фіч (`lev` від resist_plus, resist_minus, support_plus, support_minus, vol_buy, vol_sell): - **шторм** — найвищий рівень потоків; - **штиль** — обʼєднання **двох найнижчих** (див. нижче, чому обʼєднання); - **клас 2** — третій. **Чому два класи обʼєднуються в один.** Перевірка: відстань між центрами класів проти їхнього власного радіуса (корінь із середнього квадрата відстані точок до свого центра). Якщо відношення менше 1 — купки взаємно проникають і межа намальована ні на чому. У нас жодна пара з шести не дотягла до 1 (0.52…0.68), а два найнижчі класи додатково не відрізняються ні портретом, ні поведінкою ціни в межах доби. Тому вони — один стан, розрізаний навпіл. **Портрет шторму** (те, що вийшло у нас): усі шість потоків 1.4-1.9× свого масштабу, обидві стіни ТОНШІ за норму. Тобто це не «багато всього», а конкретна конфігурація: книга розсипається, а торгівля йде. ### Обовʼязкова перевірка: чи це стан ринку, чи епоха історії `python3 wclass.py confound COIN` Без зовнішньої норми найсильніший сигнал у даних — **рівень активності, який їздить від місяця до місяця**. Класифікатор радо вивчить календар замість ринку. Тому кожне твердження про ціну міряється двічі: - **абсолютно** — розмах наступної години проти медіани по всіх вікнах; - **проти того ж дня** — той самий розмах, поділений на медіану СВОГО дня. У нас це відсіяло три класи з чотирьох: ×0.83 / ×1.15 / ×0.82 зійшлися до ×0.97-0.98, тобто були мітками періоду історії. Вижив тільки шторм: ×1.78 → **×1.40**. Той самий тест на kNN-прогнозі: ρ 0.593 → 0.254, а сам денний рівень дає 0.605. **Якщо після цієї поправки клас зійшовся до 1.0 — його немає. Це календар.** --- ## 5. Крок пʼятий: як знайдено «тихе ПСА» `python3 wclass.py pcapoles COIN` Це клас з іншого конвеєра — лінії `/states`, де стани шукались у просторі **відношень до 28-денної норми**. Ми його не імпортуємо, а **перераховуємо самі на поточних даних**, бо чужий файл застаріває, а мітка мусить оновлюватись разом із датасетом. Процедура дослівно: 1. кожна фіча ділиться на власну **причинну норму за 28 діб** (ковзне середнє позаду; знаменник — кількість СПРАВЖНІХ секунд, щоб дірки не занижували); 2. перші 28 діб ідуть на прогрів і з набору випадають; 3. обрізка відношень у [1e-3, 1e4] → **логарифм** → робастний масштаб по TRAIN; 4. **PCA 32** → **UMAP** (10 компонент, 30 сусідів, min_dist 0) → **HDBSCAN** (min_cluster_size 100, метод eom); 5. полюси називаються за рівнем активності: найнижчий → **тихий**, найвищий → активний. **Нуль-контроль обовʼязковий і тут** — UMAP+HDBSCAN уміють виробляти «стійкі» класи з чистого шуму. У нас реальні дані дали 4 класи при 73% шуму, нуль — 2 класи при 99.4% шуму, тобто фактично нічого. **Спостереження, заради якого це варто робити:** полюс, який ця процедура називає «активним», у нас виявився НЕ волатильним (розмах ×0.79) — бо його метрика активності включає товщину стін. Волатильним був проміжний клас. Тому назви, що їх дає чужий конвеєр, треба перевіряти конвертом, а не приймати на віру. --- ## 6. Крок шостий: навчити детектори `python3 wclass.py clf COIN 10 storm|calm|mid|pcaquiet` **Вхід моделі — уся матриця 8 × 210 сирих значень.** Не описи з кроку 2: вони були потрібні, щоб ЗНАЙТИ класи, а детектор має працювати з сирого вікна. **Перетворення входу — `log1p(x / 0.01)` і більше нічого.** Це важливо: - логарифм потрібен, бо фічі мають важкий хвіст (одна секунда може бути в сто тисяч разів більшою за медіану), і без нього мережа бачить кілька шпильок на плоскому фоні; - `1p` — бо в даних є точні нулі, а `log(0)` це мінус нескінченність; - **стала 0.01 призначена наперед, а не взята з датасету.** Вона задає, де стоїть «коліно» логарифма — межа між лінійною і стискаючою зоною. 0.01 нижче за типові рівні всіх восьми фіч, тож стискаються всі канали. Ми **заміряли**, скільки коштує ця строгість: та сама крос-валідація з чотирма варіантами входу. Варіант з константами, вилученими з датасету, дав 0.9968; строгий `log1p(x/0.01)` — **0.9969**. Строге правило безкоштовне, тому воно й прийняте. (`log1p(x)` без сталої дає 0.9958 — гірше, бо при коліні на одиниці обʼємні фічі лишаються в лінійній зоні, де логарифм фактично вимкнений.) **Спліт:** датасет ріжеться випадково на 10 частин, у кожному прогоні 9 навчальних і 1 перевірочна, тож кожне вікно оцінене рівно один раз моделлю, яка його не бачила («out-of-fold»). Метрики — середнє по 10 фолдах ± розкид. > Застереження: випадковий спліт при корельованих сусідах дає **верхню** оцінку. Блоковий > спліт з ембарго (крок 1) дає скромніші й чесніші числа. Ми рахуємо обидва. **Два движки:** - **Conv1D** по 210 кроках: 8 каналів → 32 (ядро 7) → 64 (ядро 5) → 96 (ядро 3), після кожної згортки BatchNorm і ReLU, підвибірка вдвічі між першими двома, далі глобальне усереднення по часу, дропаут 0.2, один вихід. Втрата BCE з вагою класу, AdamW, крок 2e-3, батч 512, 8 епох. - **Градієнтний бустинг** на розгорнутому векторі 1680. Лог-регресія і kNN викинуті: перша програє, другий на сирих 1680 вимірах розсипається (евклідова відстань там міряє переважно загальний рівень шуму). **Вибір переможця:** ранжування за PR-AUC (не за AUC — цілі бувають рідкісні), поріг виставляється на частоту, що дорівнює частці цілі (тоді точність і повнота зіставні між моделями), переможець оголошується лише якщо тримається на всіх фолдах. Якщо на новій цілі порядок перевернеться — це результат, який треба показати, а не привід підкрутити. **Що вийшло у нас:** Conv1D перший на всіх чотирьох цілях і на всіх 10 фолдах. Але розрив із простими моделями росте в міру розмитості цілі — на гострому штормі лінійна модель майже наздоганяє, на розмитому серединному класі відстає помітно. Тобто гострий стан видно в рівнях, а розмитий — лише у формі. --- ## 7. Крок сьомий: продакшен і жива перевірка `python3 wclass.py final COIN storm,calm,mid,pcaquiet` · `python3 wclass_live.py COIN` Крос-валідація **не залишає моделі** — вона тренує 10 штук і викидає. Це вимірювальний прилад, а не продукт. Для роботи потрібна одна модель, тож переможець довчається окремо. **Відкладені 2% чотирма шматками.** Не 100% даних: 4 суцільні шматки по 0.5% у випадкових місцях ряду відкладаються, навколо кожного ембарго. Шматками, а не розсипом — бо сусідні вікна корелюють, і розсипана вибірка дала б завищене число. Тільки так продакшен-модель має власну вимірювану якість, а не лише успадковану від процедури. **Обидва движки йдуть у продакшен**, кожен зі своїм порогом. Розбіжність між ними — корисний сигнал: там, де вони розходяться, вікно лежить на межі класу. У нас згода на всьому датасеті: 98% на штормі, 97% на тихому ПСА і лише 90% на двох розмитих класах — незалежне підтвердження того, що межі всередині континууму умовні. **Живий контур** читає свіжі секунди, ріже вікна тим самим правилом і застосовує збережені моделі. Дві речі, які легко зіпсувати: 1. **Старти вікон мусять бути прибиті до глобальної сітки** (час кратний 210 с від епохи), а не до початку прочитаного шматка. Інакше кожен прогін ріже вікна з нового місця, вони перекриваються, і в файл лягають дублі того самого часу. 2. **Константи і поріг беруться з файлу моделі й не перераховуються.** Якщо перераховувати їх на живих даних, модель мовчки підлаштується під поточний ринок. Крок агента = вікно (210 с). --- ## 8. Що перевіряти при повторі Числа будуть інші — датасет живий. Перевіряти треба **інваріанти**: 1. gap у просторах складу не відривається від нуля; 2. силует не перевищує нуль-контроль більш ніж на 0.02; 3. Сільверман не знаходить мультимодальності в головних осях; 4. LRT відкидає K=1 **і на даних, і на нулі** (доказ, що тест тут непридатний); 5. при K ≥ 5 стійкість нуля не нижча за стійкість даних; 6. у жодної пари класів відстань між центрами не дотягує до радіуса; 7. мʼякий нуль обвалює дисперсію значно, жорсткий додає мало; 8. клас із найвищим рівнем потоків лишається помітно вище 1.0 проти того ж дня, решта сходиться до 1.0; 9. покриття правила на вікнах із довільним зсувом збігається з сітковим; 10. kNN на нуль-контролі дає ρ ≈ 0; 11. Conv1D ≥ бустинг на сирих 1680. Якщо якийсь пункт не виконався — зламався конвеєр, а не змінились дані. --- ## 9. Відкинуті напрямки (щоб не витрачати час повторно) | що | чому відкинуто | |---|---| | UMAP+HDBSCAN як основний движок класифікації | виробляє «стійкі» класи з чистого шуму; використовується лише для відтворення чужої лінії, і лише з нуль-контролем | | бутстреп-LRT та ICL як доказ існування класів | відкидають K=1 і на перемішаних даних | | рангове перетворення описів | робить маргіналії гаусовими і стирає шукану двогорбість | | багатодобова норма у ВХОДІ моделі | заборонена умовою задачі; окремо заміряно, що строгий вхід нічого не коштує | | константи масштабу, вилучені з датасету | те саме; замінені сталою, призначеною наперед | | ковзна норма (1-28 діб) для детектування полюсів | не покращує; абсолютний рівень працює краще | | лог-регресія і kNN у продакшені | програють на всіх цілях | | розділення двох найнижчих класів | геометрично одна купка; поведінка ціни однакова | | крок живого агента, більший за вікно | вікно закривається частіше, ніж агент прокидається | | читання міток із чужого файлу | застаріває; перераховуємо своєю процедурою | --- ## 10. Порядок команд ```bash python3 wclass.py data COIN # вікна + спліт з ембарго python3 wclass.py feat COIN # 89 описів вікна python3 wclass.py model COIN # табло: простори × K + нуль-контроль python3 wclass.py tests COIN # LRT, Сільверман, стійкість python3 wclass.py raws COIN # ті самі критерії на сирому векторі 1680 python3 wclass.py ladder COIN # драбина нулів: мʼякий vs жорсткий python3 wclass.py confound COIN # ВСЕ переміряти всередині доби python3 wclass.py live COIN # правило на порогах + випадкові вікна python3 wclass.py pcapoles COIN # перерахунок полюсів за стандартом /states python3 wclass.py clf COIN 10 storm # детектор: 10 фолдів × 2 движки python3 wclass.py clf COIN 10 calm python3 wclass.py clf COIN 10 mid python3 wclass.py clf COIN 10 pcaquiet python3 wclass.py final COIN storm,calm,mid,pcaquiet # продакшен + відкладені 2% python3 wclass.py scan COIN # продакшен-моделі по всій історії python3 wclass.py report COIN full 16 4 # звіт + графіки python3 wclass_live.py COIN # живий контур (агент кожні 210 с) ``` Змінна `WCLASS_PREP` (`fixed` | `pure` | `asinh` | `scale`) перемикає варіант входу для повторення заміру з кроку 6. --- ## 11. Повний код Нижче — усі файли лінії цілком. Цього достатньо, щоб відтворити конвеєр з нуля. ### Головний скрипт лінії `wclass.py` — усі етапи: data · feat · model · tests · raws · ladder · confound · live · pcapoles · clf · final · scan · report ```python """ СТАТИСТИЧНО КОРЕКТНА КЛАСИФІКАЦІЯ ВІКОН 210 с (/wclass). Обмеження задачі (задане користувачем): класифікуємо вікно з 8 фіч × 210 с, і вільно змінювати можна ЛИШЕ те, що робиться ВСЕРЕДИНІ вікна. Тобто: · довжина вікна 210 с і набір з 8 фіч — фіксовані; · жодної інформації ззовні вікна: ні багатодобової норми, ні ціни, ні часу, ні сусідніх вікон. Вхід кожного вікна — рівно його 1680 сирих чисел. · єдиний виняток, і він статистично необхідний: КОНСТАНТИ масштабування (квантилі, медіани, PCA, параметри моделі) оцінюються на TRAIN і однакові для всіх вікон. Це властивість вибірки, а не підказка про конкретне вікно. Що робимо і чому саме так: 1. ДАНІ. Дедуплікація, суцільна 1-секундна шкала (дірки — лінійна інтерполяція з маскою), вікна стик у стик по 210 с, вікно приймається при ≥95% справжніх секунд. Спліт блоками по 2 год з ембарго 30 хв: сусідні вікна корелюють, тому випадковий спліт по вікнах завищив би все. 2. ОПИС ВІКНА (єдине місце свободи). Сирий ряд фічі — невідʼємний, з нулями і важким хвостом; середнє по ньому ламає один тик. Тому опис — набір статистик, у яких відомо, що вони міряють: рівень (лог), розкид, концентрація (частка суми у 10 найбільших секундах), частка тихих секунд, форма (2 коефіцієнти Лежандра — нахил і вигин), інерція (автокореляція), і КОМПОЗИЦІЯ вікна — clr-перетворення 8 нормованих мас (правильна алгебра для часток). ТРИ ПОДАННЯ, щоб питання «класи — це склад ринку чи просто рівень активності?» закривалось числом: full (усе), free (без абсолютних рівнів, інваріантне до множення вікна на константу), lev (лише рівні). 3. МОДЕЛЬ. Скінченна суміш гаусіан (GMM) — на відміну від UMAP+HDBSCAN має правдоподібність, тож K обирається статистикою, а не оком. Критерій — ПРАВДОПОДІБНІСТЬ НА ВІДКЛАДЕНИХ вікнах, плюс BIC та ICL (ICL карає перекриття компонент: якщо це континуум, ICL зупиниться на K=1). ВАЖЛИВО: правдоподібність зіставна лише В МЕЖАХ одного простору — різні подання порівнюються не нею, а відривом від нуль-контролю і стійкістю. 4. ПЕРЕВІРКИ (без них це не класифікація, а картинка): · параметричний bootstrap-LRT K проти K+1 — чи взагалі є більше однієї компоненти (звичайний χ² тут не працює, межа простору параметрів); · НУЛЬ-КОНТРОЛЬ: клітинки (фіча, секунда) перемішуються МІЖ вікнами — гине все, крім маргіналій; увесь конвеєр переганяється заново; · стійкість: ARI між мітками двох ПОВНИХ перебудов на непересічних часових половинах, зіставлених на спільній відкладеній вибірці; · тест Сільвермана на мультимодальність у дискримінантній проєкції. 5. КОНВЕРТ (post hoc, після кластеризації): forward-волатильність, календар (чи не є «клас» просто шматком історії), збіг з /events і /standard. 6. КОНФАУНД ЕПОХИ. Без зовнішньої норми найсильніший сигнал у даних — це рівень активності, який їздить від місяця до місяця. Тому кожне твердження переміряється ВСЕРЕДИНІ ДОБИ (forward-розмах ділиться на медіану того ж дня). Що не пережило цю перевірку — то був календар, а не стан ринку. Етапи: python3 wclass.py data BTC # сирі вікна 210 с + спліт з ембарго python3 wclass.py feat BTC # 89 внутрішньовіконних описів python3 wclass.py model BTC # табло: 6 просторів × K=1..12 + 5 нулів python3 wclass.py tests BTC # LRT, Сільверман, стійкість python3 wclass.py raws BTC # ті самі критерії на сирому векторі 1680 python3 wclass.py ladder BTC # драбина нулів: мʼякий vs жорсткий python3 wclass.py confound BTC # перевірка всередині доби python3 wclass.py report BTC full 16 4 """ import os, sys, json, time import numpy as np import standard_signals as ST from standard_signals import FEATS ROOT = os.path.dirname(os.path.abspath(__file__)) OUT_DIR = os.path.join(ROOT, "media", "analyst", "wclass") CACHE = os.path.join(OUT_DIR, "cache") WIN = 210 # вікно, секунд — ФІКСОВАНЕ умовою задачі STEP = WIN # вікна стик у стик MIN_REAL = 0.95 # мінімум справжніх (не інтерпольованих) секунд у вікні BLOCK_H = 2 # блок спліту, годин EMBARGO_S = 1800 # ембарго навколо валідації, секунд VAL_FRAC = 0.10 SEED = 20260804 def log(*a): print(time.strftime("[%H:%M:%S]"), *a, flush=True) # ============================ ЕТАП 1: дані ============================ def build_data(coin): """Сирі вікна. Жодного ділення на норму — рівно те, що є всередині вікна.""" os.makedirs(CACHE, exist_ok=True) log(f"{coin}: читаю датасет…") S = ST.load(coin) if S is None: log("немає даних"); return None n_raw = len(S["ts"]) _, first = np.unique(S["ts"], return_index=True) first = np.sort(first) n_dup = n_raw - len(first) for k in ["ts", "price"] + FEATS: S[k] = S[k][first] ts = S["ts"] log(f"{coin}: {n_raw} рядків → {len(ts)} після дедуплікації (−{n_dup})") t0, t1 = int(ts[0]), int(ts[-1]) N = t1 - t0 + 1 pos = (ts - t0).astype(np.int64) present = np.zeros(N, bool) present[pos] = True n_miss = int(N - present.sum()) log(f"{coin}: шкала {N} с ({N/86400:.1f} діб) · відсутніх секунд {n_miss} " f"({n_miss/N*100:.2f}%) — інтерполюю") gaps = np.diff(ts) big = np.flatnonzero(gaps > 60) holes = sorted(({"t": int(ts[i]), "sec": int(gaps[i] - 1)} for i in big), key=lambda h: -h["sec"])[:20] grid_i = np.arange(N, dtype=np.float64) px = np.interp(grid_i, pos, S["price"]).astype(np.float32) # ЗАМОРОЖЕННЯ МЕЖІ ДАНИХ: WCLASS_END= обрізає ряд, щоб # повторний прогін через місяць рахувався на ТОМУ Ж наборі вікон. Без нього # датасет росте щодня і всі частки/кількості законно їдуть. end = os.environ.get("WCLASS_END", "").strip() if end: import datetime as _dt te = int(end) if end.isdigit() else int( _dt.datetime.strptime(end, "%Y-%m-%d").replace(tzinfo=_dt.timezone.utc).timestamp()) if te < t1: N = min(N, te - t0 + 1) log(f"{coin}: межу даних заморожено на {end} → {N} с ({N/86400:.1f} діб)") starts = np.arange(0, N - WIN + 1, STEP, dtype=np.int64) cpres = np.concatenate([[0], np.cumsum(present, dtype=np.int64)]) real = (cpres[starts + WIN] - cpres[starts]) / WIN keep = real >= MIN_REAL starts = starts[keep]; real = real[keep].astype(np.float32) n_win = len(starts) log(f"{coin}: вікон {n_win} (відсіяно за повнотою {int((~keep).sum())})") if n_win < 2000: log("замало вікон"); return None idx = (starts[:, None] + np.arange(WIN, dtype=np.int64)[None, :]) X = np.empty((n_win, 8, WIN), np.float32) for j, f in enumerate(FEATS): v = np.interp(grid_i, pos, S[f].astype(np.float64)).astype(np.float32) X[:, j, :] = v[idx] del v log(f" {f}: нарізано") del grid_i, idx ok = np.isfinite(X).all(axis=(1, 2)) if not ok.all(): log(f"{coin}: вікон з нескінченністю {int((~ok).sum())} — геть") X = X[ok]; starts = starts[ok]; real = real[ok]; n_win = len(starts) wts = (t0 + starts).astype(np.int64) wpx = px[starts] # ціна на кінець вікна і через годину — ТІЛЬКИ для post hoc конверта nxt = np.minimum(starts + WIN, N - 1) fwd_end = np.minimum(starts + WIN + 3600, N - 1) px_end = px[nxt] px_fwd = px[fwd_end] # розмах ціни в наступну годину (конверт, не вхід моделі) rng_fwd = np.zeros(n_win, np.float32) for i in range(0, n_win, 5000): sl = slice(i, min(i + 5000, n_win)) a = starts[sl] + WIN; b = np.minimum(a + 3600, N) rng_fwd[sl] = [(px[x:y].max() - px[x:y].min()) / max(px[x], 1e-9) * 100 if y > x else 0.0 for x, y in zip(a, b)] rng = np.random.default_rng(SEED) blk = ((wts - wts[0]) // (BLOCK_H * 3600)).astype(np.int64) ub = np.unique(blk) sh = rng.permutation(len(ub)) n_val_b = max(1, int(round(len(ub) * VAL_FRAC))) val_blocks = set(ub[sh[:n_val_b]].tolist()) is_val = np.array([b in val_blocks for b in blk]) val = np.flatnonzero(is_val).astype(np.int32) vts = wts[val] near = np.zeros(len(wts), bool) j = np.searchsorted(vts, wts) for shift in (0, -1): k = np.clip(j + shift, 0, len(vts) - 1) near |= np.abs(wts - vts[k]) <= EMBARGO_S train = np.flatnonzero(~is_val & ~near).astype(np.int32) n_emb = int((~is_val & near).sum()) log(f"{coin}: навчання {len(train)} · валідація {len(val)} · ембарго {n_emb}") np.savez(os.path.join(CACHE, f"{coin}_data.npz"), X=X, ts=wts, px=wpx, px_end=px_end, px_fwd=px_fwd, rng_fwd=rng_fwd, real=real, train=train, val=val) meta = {"coin": coin, "built": int(time.time()), "rows_raw": int(n_raw), "rows_dedup": int(len(ts)), "dup": int(n_dup), "t0": t0, "t1": t1, "days": round(N / 86400, 1), "grid_sec": int(N), "missing_sec": n_miss, "missing_pct": round(n_miss / N * 100, 3), "holes_top": holes, "win": WIN, "step": STEP, "min_real": MIN_REAL, "n_win": int(n_win), "n_train": int(len(train)), "n_val": int(len(val)), "n_embargo": n_emb, "block_h": BLOCK_H, "embargo_s": EMBARGO_S, "seed": SEED, "real_mean": round(float(real.mean()), 5)} with open(os.path.join(CACHE, f"{coin}_data.json"), "w") as fh: json.dump(meta, fh, ensure_ascii=False) log(f"{coin}: дані збережено · X={X.shape}") return meta # ==================== ЕТАП 2: опис вікна ==================== # Єдине місце, де умова задачі лишає свободу. Кожна статистика — функція # ВИКЛЮЧНО 1680 чисел вікна; константи масштабу s_f беруться з TRAIN. def _legendre(win=WIN): """Ортонормовані поліноми 0/1/2 на сітці вікна (через QR — точна ортогональність).""" t = np.linspace(-1.0, 1.0, win) V = np.stack([np.ones(win), t, t * t], axis=1) Q, _ = np.linalg.qr(V) return Q[:, 1].astype(np.float32), Q[:, 2].astype(np.float32) # нахил, вигин P1, P2 = _legendre() TOPK = 10 # скільки найбільших секунд у мірі концентрації QUIET_U = 0.05 # «тиха» секунда: менш ніж 5% від типового рівня фічі def descriptors(Xc, s): """Xc (m,8,210) сирі → (m, 65) описів. s (8,) — масштаби фіч з TRAIN.""" m = Xc.shape[0] u = Xc / s[None, :, None] # безрозмірне L = np.log1p(u) # рівень у лог-одиницях lev = L.mean(axis=2) sd = L.std(axis=2) tot = u.sum(axis=2) top = np.partition(u, -TOPK, axis=2)[:, :, -TOPK:].sum(axis=2) conc = top / np.maximum(tot, 1e-12) # концентрація суми у 10 с quiet = (u < QUIET_U).mean(axis=2) # частка тихих секунд trend = L @ P1 curv = L @ P2 a = L[:, :, :-1]; b = L[:, :, 1:] am = a.mean(axis=2, keepdims=True); bm = b.mean(axis=2, keepdims=True) num = ((a - am) * (b - bm)).mean(axis=2) den = a.std(axis=2) * b.std(axis=2) ac1 = np.where(den > 1e-9, num / np.maximum(den, 1e-12), 0.0) # композиція вікна: clr від 8 нормованих мас (правильна алгебра часток) lm = np.log(tot + 1e-6) size = lm.mean(axis=1, keepdims=True) # «розмір» вікна (геом. середнє мас) clr = lm - size # склад, інваріантний до розміру # похідні форми, незалежні від амплітуди cv = sd / np.maximum(np.abs(lev), 1e-6) trend_n = trend / np.maximum(sd, 1e-6) curv_n = curv / np.maximum(sd, 1e-6) D = np.concatenate([lev, sd, conc, quiet, trend, curv, ac1, cv, trend_n, curv_n, clr, size], axis=1) return D.astype(np.float32) def desc_names(): n = [] for g in ["lev", "sd", "conc", "quiet", "trend", "curv", "ac1", "cv", "trendN", "curvN", "clr"]: n += [f"{g}:{f}" for f in FEATS] return n + ["size"] NAMES = desc_names() # три подання (питання «класи — це склад чи рівень активності?») REPR = { "full": ["lev", "sd", "conc", "quiet", "trend", "curv", "ac1", "clr", "size"], "free": ["conc", "cv", "trendN", "curvN", "ac1", "clr"], # без абсолютних рівнів "lev": ["lev", "size"], } def repr_cols(name): grp = REPR[name] return np.array([i for i, n in enumerate(NAMES) if n.split(":")[0] in grp or n in grp]) def build_feat(coin, chunk=4000): d = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") X = d["X"]; train = np.asarray(d["train"]) n = X.shape[0] # масштаб фічі — середнє по TRAIN (константа вибірки, не підказка про вікно) acc = np.zeros(8, np.float64); cnt = 0 for i in range(0, len(train), chunk): sl = np.asarray(train[i:i + chunk]) Xc = np.asarray(X[sl]) acc += Xc.mean(axis=(0, 2), dtype=np.float64) * len(sl); cnt += len(sl) s = (acc / cnt).astype(np.float32) log(f"{coin}: масштаби фіч (TRAIN) " + " ".join(f"{f}={v:.4g}" for f, v in zip(FEATS, s))) D = np.empty((n, len(NAMES)), np.float32) for i in range(0, n, chunk): D[i:i + chunk] = descriptors(np.asarray(X[i:i + chunk]), s) if (i // chunk) % 5 == 0: log(f" описи {i}/{n}") bad = ~np.isfinite(D) if bad.any(): log(f"{coin}: нескінченностей в описах {int(bad.sum())} → 0") D[bad] = 0.0 np.savez(os.path.join(CACHE, f"{coin}_feat.npz"), D=D, scale=s, names=np.array(NAMES)) log(f"{coin}: описи готові {D.shape}") return D # ==================== простір: вінзоризація → робастний z → PCA ==================== # Усе підганяється ТІЛЬКИ на TRAIN. Рангове перетворення навмисно НЕ беремо: # воно робить кожну маргіналію гаусовою і разом з важким хвостом стирає саме ту # двогорбість, яку ми шукаємо. Вінзоризація хвости приборкує, форму лишає. WINSOR = (0.001, 0.999) class Space: def __init__(self, cols, d): self.cols = cols; self.d = d def fit(self, D, train): from sklearn.decomposition import PCA A = D[:, self.cols] T = A[train] self.lo = np.quantile(T, WINSOR[0], axis=0) self.hi = np.quantile(T, WINSOR[1], axis=0) T = np.clip(T, self.lo, self.hi) self.med = np.median(T, axis=0) q75, q25 = np.percentile(T, [75, 25], axis=0) self.iqr = np.where(q75 - q25 > 1e-9, q75 - q25, 1.0) Z = (T - self.med) / self.iqr self.pca = PCA(n_components=min(self.d, Z.shape[1]), whiten=True, random_state=SEED).fit(Z) self.evr = float(self.pca.explained_variance_ratio_.sum()) return self def transform(self, D): A = np.clip(D[:, self.cols], self.lo, self.hi) return self.pca.transform((A - self.med) / self.iqr).astype(np.float32) def load_all(coin): f = np.load(os.path.join(CACHE, f"{coin}_feat.npz")) d = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") return {"D": f["D"], "scale": f["scale"], "train": np.asarray(d["train"]), "val": np.asarray(d["val"]), "ts": np.asarray(d["ts"]), "px": np.asarray(d["px"]), "rng_fwd": np.asarray(d["rng_fwd"])} # ==================== НУЛЬ-КОНТРОЛЬ ==================== def null_series(coin, seed, chunk=4000): """МʼЯКШИЙ нуль: перемішуються цілі ряди фіч між вікнами (кожна фіча своєю перестановкою). Форма кожної фічі всередині вікна ЗБЕРІГАЄТЬСЯ, гине лише спільність фіч у вікні. Різниця з жорстким нулем показує, де саме структура.""" d = np.load(os.path.join(CACHE, f"{coin}_data.npz")) X = d["X"]; train = np.asarray(d["train"]); val = np.asarray(d["val"]) rng = np.random.default_rng(seed) for part in (train, val): Xp = X[part] for j in range(8): Xp[:, j, :] = Xp[rng.permutation(len(part)), j, :] X[part] = Xp del Xp s = np.load(os.path.join(CACHE, f"{coin}_feat.npz"))["scale"] D = np.empty((X.shape[0], len(NAMES)), np.float32) for i in range(0, X.shape[0], chunk): D[i:i + chunk] = descriptors(X[i:i + chunk], s) D[~np.isfinite(D)] = 0.0 del X return D def null_desc(coin, seed, chunk=4000): """Клітинки (фіча, секунда) перемішуються МІЖ вікнами — окремо в TRAIN і VAL, щоб не було витоку. Гине все, крім маргінальних розподілів. Далі — той самий конвеєр описів. Усе, що конвеєр знайде тут, він вигадав.""" d = np.load(os.path.join(CACHE, f"{coin}_data.npz")) X = d["X"]; train = np.asarray(d["train"]); val = np.asarray(d["val"]) rng = np.random.default_rng(seed) for part in (train, val): Xp = X[part] for j in range(8): for t in range(WIN): Xp[:, j, t] = Xp[rng.permutation(len(part)), j, t] X[part] = Xp del Xp f = np.load(os.path.join(CACHE, f"{coin}_feat.npz")) s = f["scale"] D = np.empty((X.shape[0], len(NAMES)), np.float32) for i in range(0, X.shape[0], chunk): D[i:i + chunk] = descriptors(X[i:i + chunk], s) D[~np.isfinite(D)] = 0.0 del X return D # ==================== ГІЛКА А: числові методи на відстанях ==================== def kmeans_logw(Z, ks, seed=SEED, n_init=6): """log сумарної внутрішньокластерної дисперсії (Тібширані W_K) для кожного K.""" from sklearn.cluster import KMeans out = {} for K in ks: if K == 1: out[K] = (float(np.log(((Z - Z.mean(0)) ** 2).sum())), None) continue km = KMeans(n_clusters=K, n_init=n_init, random_state=seed).fit(Z) out[K] = (float(np.log(max(km.inertia_, 1e-12))), km) return out def nn_stats(Z, m=4000, seed=SEED): """Відстаневі діагностики: середня відстань до 1-го і 10-го сусіда, відносний розкид попарних відстаней (континуум → малий розкид).""" from sklearn.neighbors import NearestNeighbors rng = np.random.default_rng(seed) s = Z[rng.choice(len(Z), min(m, len(Z)), replace=False)] nn = NearestNeighbors(n_neighbors=11).fit(s) dist, _ = nn.kneighbors(s) from scipy.spatial.distance import pdist pd = pdist(s[:2000]) return {"nn1": float(dist[:, 1].mean()), "nn10": float(dist[:, 10].mean()), "pd_mean": float(pd.mean()), "pd_cv": float(pd.std() / pd.mean())} def silhouette(Z, lab, m=5000, seed=SEED): from sklearn.metrics import silhouette_score rng = np.random.default_rng(seed) i = rng.choice(len(Z), min(m, len(Z)), replace=False) if len(np.unique(lab[i])) < 2: return float("nan") return float(silhouette_score(Z[i], lab[i])) # ---- тест Сільвермана на мультимодальність (непараметричний) ---- def _n_modes(x, h, grid=512): g = np.linspace(x.min() - 3 * h, x.max() + 3 * h, grid) z = (g[:, None] - x[None, :]) / h dens = np.exp(-0.5 * z * z).sum(axis=1) d = np.diff(dens) return int(((d[:-1] > 0) & (d[1:] <= 0)).sum()) def critical_bw(x, k, lo=1e-4, hi=None, it=40): """Найменша ширина ядра, за якої мод не більше k.""" sd = x.std() hi = hi if hi is not None else 3 * sd for _ in range(it): mid = 0.5 * (lo + hi) if _n_modes(x, mid) > k: lo = mid else: hi = mid return hi def silverman_test(x, k=1, B=100, seed=SEED, m=2000): """H0: щільність має не більше k мод. Згладжений бутстреп при критичній ширині.""" rng = np.random.default_rng(seed) x = np.asarray(x, np.float64) if len(x) > m: x = x[rng.choice(len(x), m, replace=False)] x = (x - x.mean()) / x.std() h = critical_bw(x, k) sd = x.std(); n = len(x) cnt = 0 for _ in range(B): idx = rng.integers(0, n, n) y = x[idx] + h * rng.standard_normal(n) y = x.mean() + (y - y.mean()) / np.sqrt(1 + h * h / (sd * sd)) if _n_modes(y, h) > k: cnt += 1 return {"h_crit": float(h), "p": float((cnt + 1) / (B + 1)), "n_boot": B} # ==================== ГІЛКА Б: суміш гаусіан ==================== def gmm_fit(Ztr, K, cov="full", seed=SEED, n_init=3, reg=1e-4): from sklearn.mixture import GaussianMixture return GaussianMixture(n_components=K, covariance_type=cov, n_init=n_init, reg_covar=reg, random_state=seed, max_iter=300).fit(Ztr) def gmm_scores(g, Ztr, Zval): """BIC/ICL на TRAIN + правдоподібність на ВІДКЛАДЕНИХ вікнах.""" bic = float(g.bic(Ztr)) R = np.nan_to_num(g.predict_proba(Ztr), nan=0.0) ent = float(-np.where(R > 1e-12, R * np.log(np.maximum(R, 1e-300)), 0.0).sum()) return {"bic": bic, "icl": bic + 2 * ent, "ent": ent, "val_ll": float(g.score(Zval)), "train_ll": float(g.score(Ztr)), "maxpost": float(R.max(axis=1).mean())} def boot_lrt(Ztr, K, cov="full", B=40, n=8000, seed=SEED): """Параметричний бутстреп-LRT: K проти K+1. Звичайний χ² тут не діє — при H0 параметри зайвої компоненти лежать на межі простору.""" rng = np.random.default_rng(seed) idx = rng.choice(len(Ztr), min(n, len(Ztr)), replace=False) Z = Ztr[idx] g0 = gmm_fit(Z, K, cov, n_init=2) g1 = gmm_fit(Z, K + 1, cov, n_init=2) obs = 2 * len(Z) * (g1.score(Z) - g0.score(Z)) null = [] for b in range(B): Y, _ = g0.sample(len(Z)) Y = Y.astype(np.float32) try: a = gmm_fit(Y, K, cov, seed=SEED + b, n_init=1) c = gmm_fit(Y, K + 1, cov, seed=SEED + b, n_init=1) null.append(2 * len(Y) * (c.score(Y) - a.score(Y))) except Exception: pass null = np.array(null) p = float((1 + (null >= obs).sum()) / (len(null) + 1)) return {"K": K, "lr": float(obs), "null_med": float(np.median(null)) if len(null) else None, "null_q95": float(np.quantile(null, 0.95)) if len(null) else None, "p": p, "B": int(len(null))} # ==================== ЕТАП 3: табло ==================== KS = list(range(1, 13)) SPACES = [("full", 8), ("full", 16), ("free", 8), ("free", 16), ("lev", 4), ("lev", 8)] N_NULL = 5 def evaluate(D, train, val, rp, d, ks=KS, with_gmm=True, seed=SEED): """Одне подання × одна розмірність: числові критерії + GMM. Простір підганяється НАНОВО на train (жодного повторного використання).""" sp = Space(repr_cols(rp), d).fit(D, train) Ztr = sp.transform(D[train]); Zva = sp.transform(D[val]) res = {"repr": rp, "d": d, "evr": round(sp.evr, 4), "nn": nn_stats(Ztr, seed=seed)} kw = kmeans_logw(Ztr, ks, seed=seed) res["logw"] = {K: round(kw[K][0], 5) for K in ks} res["sil"] = {K: (round(silhouette(Ztr, kw[K][1].labels_, seed=seed), 4) if K > 1 else None) for K in ks} if with_gmm: res["gmm"] = {} for cov in ("diag", "full"): cur = {} for K in ks: try: g = gmm_fit(Ztr, K, cov, seed=seed, n_init=2) s = gmm_scores(g, Ztr, Zva) cur[K] = {k: round(v, 4) for k, v in s.items()} except Exception as e: cur[K] = {"err": str(e)[:80]} res["gmm"][cov] = cur return res def model(coin, n_null=N_NULL): os.makedirs(OUT_DIR, exist_ok=True) A = load_all(coin) D, train, val = A["D"], A["train"], A["val"] out = {"coin": coin, "built": int(time.time()), "n_win": int(len(D)), "n_train": int(len(train)), "n_val": int(len(val)), "ks": KS, "spaces": [], "null": []} for rp, d in SPACES: t0 = time.time() r = evaluate(D, train, val, rp, d) out["spaces"].append(r) log(f"{coin}: {rp}/d{d} готово за {time.time()-t0:.0f}с · evr={r['evr']:.3f} " f"· sil(2)={r['sil'][2]} · pd_cv={r['nn']['pd_cv']:.3f}") for b in range(n_null): t0 = time.time() Dn = null_desc(coin, seed=1000 + b) rep = [] for rp, d in SPACES: rep.append(evaluate(Dn, train, val, rp, d, seed=SEED + b)) out["null"].append(rep) log(f"{coin}: нуль-контроль {b+1}/{n_null} за {time.time()-t0:.0f}с") del Dn with open(os.path.join(CACHE, f"{coin}_model.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: табло збережено") return out # ==================== ЕТАП 4: перевірки ==================== def fisher_dir(Z, lab): """Напрямок найбільшого розділення двох груп (Фішер) — рахується на TRAIN.""" a = Z[lab == 0]; b = Z[lab == 1] Sw = np.cov(a.T) * len(a) + np.cov(b.T) * len(b) w = np.linalg.solve(Sw + np.eye(Z.shape[1]) * 1e-6, b.mean(0) - a.mean(0)) return w / np.linalg.norm(w) def stability(D, train, val, rp, d, K, seed=SEED): """ARI між двома ПОВНИМИ перебудовами конвеєра на непересічних часових половинах TRAIN, зіставленими на спільній валідації.""" from sklearn.cluster import KMeans from sklearn.metrics import adjusted_rand_score h = len(train) // 2 labs = [] for part in (train[:h], train[h:]): sp = Space(repr_cols(rp), d).fit(D, part) Zp = sp.transform(D[part]); Zv = sp.transform(D[val]) km = KMeans(n_clusters=K, n_init=6, random_state=seed).fit(Zp) labs.append(km.predict(Zv)) return float(adjusted_rand_score(labs[0], labs[1])) def tests(coin, rp="full", d=16, kmax=6, B_lrt=40, B_sil=100, n_null=3): A = load_all(coin) D, train, val = A["D"], A["train"], A["val"] out = {"coin": coin, "repr": rp, "d": d, "built": int(time.time())} sp = Space(repr_cols(rp), d).fit(D, train) Ztr = sp.transform(D[train]); Zva = sp.transform(D[val]) # 1. параметричний бутстреп-LRT K vs K+1 log("bootstrap-LRT…") out["lrt"] = [] for K in range(1, kmax + 1): r = boot_lrt(Ztr, K, "full", B=B_lrt) out["lrt"].append(r) log(f" K={K}→{K+1}: LR={r['lr']:.0f} нуль(медіана/95%)=" f"{r['null_med']:.0f}/{r['null_q95']:.0f} p={r['p']:.3f}") # 2. мультимодальність: перші 5 головних осей + напрямок Фішера (out-of-sample) log("тест Сільвермана…") from sklearn.cluster import KMeans km2 = KMeans(n_clusters=2, n_init=6, random_state=SEED).fit(Ztr) w = fisher_dir(Ztr, km2.labels_) out["silverman"] = {} for name, x in [(f"PC{i+1}", Zva[:, i]) for i in range(min(5, d))] + \ [("fisher2", Zva @ w)]: out["silverman"][name] = silverman_test(x, k=1, B=B_sil) log(f" {name}: p={out['silverman'][name]['p']:.3f}") # 3. стійкість повної перебудови + той самий тест на нуль-контролі log("стійкість…") out["stab"] = {K: round(stability(D, train, val, rp, d, K), 4) for K in range(2, kmax + 1)} log(f" реальні дані: {out['stab']}") ns = {K: [] for K in range(2, kmax + 1)} nsil = {} for b in range(n_null): Dn = null_desc(coin, seed=2000 + b) for K in range(2, kmax + 1): ns[K].append(stability(Dn, train, val, rp, d, K, seed=SEED + b)) if b == 0: spn = Space(repr_cols(rp), d).fit(Dn, train) Zn = spn.transform(Dn[train]); Zvn = spn.transform(Dn[val]) kn = KMeans(n_clusters=2, n_init=6, random_state=SEED).fit(Zn) wn = fisher_dir(Zn, kn.labels_) nsil = {"fisher2": silverman_test(Zvn @ wn, k=1, B=B_sil), "PC1": silverman_test(Zvn[:, 0], k=1, B=B_sil)} out["lrt_null"] = boot_lrt(Zn, 1, "full", B=B_lrt) del Dn log(f" нуль {b+1}/{n_null}") out["stab_null"] = {K: round(float(np.mean(v)), 4) for K, v in ns.items()} out["silverman_null"] = nsil log(f" нуль-контроль: {out['stab_null']}") with open(os.path.join(CACHE, f"{coin}_tests.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log("перевірки збережено") return out # ==================== ЕТАП 5: конверт (post hoc) ==================== def gap_curve(mj, rp, d): """Gap-статистика Тібширані з нуль-контролем як еталоном (замість рівномірного куба: наш нуль зберігає маргіналії, тому це чесніший еталон).""" real = next(s for s in mj["spaces"] if s["repr"] == rp and s["d"] == d) nulls = [next(s for s in rep if s["repr"] == rp and s["d"] == d) for rep in mj["null"]] ks = mj["ks"] out = [] for K in ks: r = real["logw"][str(K)] if str(K) in real["logw"] else real["logw"][K] nv = np.array([(n["logw"][str(K)] if str(K) in n["logw"] else n["logw"][K]) for n in nulls]) sk = float(nv.std(ddof=1) * np.sqrt(1 + 1 / len(nv))) out.append({"K": K, "logw": r, "null_mean": float(nv.mean()), "gap": float(nv.mean() - r), "sk": sk}) # правило Тібширані: найменше K, де gap(K) ≥ gap(K+1) − s_{K+1} kstar = None for i in range(len(out) - 1): if out[i]["gap"] >= out[i + 1]["gap"] - out[i + 1]["sk"]: kstar = out[i]["K"]; break return out, kstar def cramers_v(lab, grp): from scipy.stats import chi2_contingency tab = np.zeros((lab.max() + 1, int(grp.max()) + 1)) for a, b in zip(lab, grp.astype(int)): tab[a, b] += 1 tab = tab[:, tab.sum(0) > 0] chi2, p, _, _ = chi2_contingency(tab) n = tab.sum() return float(np.sqrt(chi2 / (n * (min(tab.shape) - 1)))), float(p) def subspace_angles(D, train, rp, d): """Наскільки ті самі осі знаходяться на різних половинах історії.""" from scipy.linalg import subspace_angles as sa h = len(train) // 2 B = [] for part in (train[:h], train[h:]): sp = Space(repr_cols(rp), d).fit(D, part) B.append(sp.pca.components_.T) ang = sa(B[0], B[1]) return [round(float(np.cos(a)), 4) for a in ang] def report(coin, rp="full", d=16, K=4): import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt from sklearn.cluster import KMeans from scipy.stats import mannwhitneyu, kruskal os.makedirs(OUT_DIR, exist_ok=True) A = load_all(coin) D, train, val, ts = A["D"], A["train"], A["val"], A["ts"] mj = json.load(open(os.path.join(CACHE, f"{coin}_model.json"))) tj = {} p = os.path.join(CACHE, f"{coin}_tests.json") if os.path.exists(p): tj = json.load(open(p)) extra = {} for nm in ("raw", "ladder", "confound", "live", "learn"): q = os.path.join(CACHE, f"{coin}_{nm}.json") if os.path.exists(q): extra[nm] = json.load(open(q)) # усі навчені цілі: {coin}_clf.json (шторм) + {coin}_clf_<ціль>.json import glob as _glob targets = [("clf", "")] for q in sorted(_glob.glob(os.path.join(CACHE, f"{coin}_clf_*.json"))): t = os.path.basename(q)[len(coin) + 5:-5] targets.append((f"clf_{t}", f"_{t}")) for nm, sfx in targets: q = os.path.join(CACHE, f"{coin}_clf{sfx}.json") if os.path.exists(q): extra[nm] = json.load(open(q)) # криві точність/повнота з out-of-fold прогнозів (кожне вікно оцінене раз) from sklearn.metrics import precision_recall_curve, roc_curve for nm, sfx in targets: qo = os.path.join(CACHE, f"{coin}_clf_oof{sfx}.npz") if nm not in extra or not os.path.exists(qo): continue z = np.load(qo); yv = z["y"] cur = {} for k in extra[nm]["models"]: if k not in z: continue pv = z[k]; ok = np.isfinite(pv) pr, rc, _ = precision_recall_curve(yv[ok], pv[ok]) fp, tp, _ = roc_curve(yv[ok], pv[ok]) gi = np.linspace(0, len(rc) - 1, 80).astype(int) gj = np.linspace(0, len(fp) - 1, 80).astype(int) cur[k] = {"rc": [round(float(v), 4) for v in rc[gi]], "pr": [round(float(v), 4) for v in pr[gi]], "fpr": [round(float(v), 4) for v in fp[gj]], "tpr": [round(float(v), 4) for v in tp[gj]]} extra[nm]["curves"] = cur # прогноз найкращої моделі на кожне вікно (out-of-fold) — для графіка bk = extra[nm].get("best") if bk and bk in z: sub = z["sub"] if "sub" in z.files else np.arange(len(z[bk]), dtype=np.int64) ser = np.full(len(D), np.nan, np.float32) ser[sub] = z[bk] extra[nm]["oof"] = {"model": bk, "p": [None if not np.isfinite(v) else int(round(float(v) * 100)) for v in ser]} out = {"coin": coin, "built": int(time.time()), "repr": rp, "d": d, "K": K, "n_win": int(len(D)), "n_train": int(len(train)), "n_val": int(len(val)), "win": WIN, "tests": tj, **extra} out["gap"] = {} for r_, d_ in SPACES: g, ks_ = gap_curve(mj, r_, d_) out["gap"][f"{r_}/{d_}"] = {"curve": g, "kstar": ks_} out["angles"] = subspace_angles(D, train, rp, d) sp = Space(repr_cols(rp), d).fit(D, train) Z = sp.transform(D) # що саме означають осі: найбільші внески описів nm = [NAMES[i] for i in sp.cols] out["axes"] = [] for a in range(min(3, d)): w = sp.pca.components_[a] o = np.argsort(-np.abs(w))[:8] out["axes"].append({"axis": a + 1, "evr": round(float(sp.pca.explained_variance_ratio_[a]), 4), "top": [[nm[i], round(float(w[i]), 3)] for i in o]}) km = KMeans(n_clusters=K, n_init=10, random_state=SEED).fit(Z[train]) lab = km.predict(Z) out["shares"] = [round(float((lab == k).mean()), 4) for k in range(K)] # «штормовий» клас визначаємо ДЕТЕРМІНОВАНО з боку ВХОДУ (найвищий сумарний # рівень 6 потокових фіч), а не за номером k-means і не за майбутньою ціною flow_i = [NAMES.index(f"lev:{f}") for f in FEATS if f not in ("const_resist", "const_support")] fl = np.array([float(np.median(D[lab == k][:, flow_i].sum(axis=1))) for k in range(K)]) out["storm_cls"] = int(np.argmax(fl)) out["flow_level"] = [round(float(v), 3) for v in fl] # ---- портрети класів у сирих одиницях (медіана рівня u = x / масштаб) ---- lev_i = [i for i, n in enumerate(NAMES) if n.startswith("lev:")] clr_i = [i for i, n in enumerate(NAMES) if n.startswith("clr:")] con_i = [i for i, n in enumerate(NAMES) if n.startswith("conc:")] out["portrait"] = [] for k in range(K): m = lab == k out["portrait"].append({ "cls": k, "n": int(m.sum()), "lev": {f: round(float(np.expm1(np.median(D[m, i]))), 4) for f, i in zip(FEATS, lev_i)}, "clr": {f: round(float(np.median(D[m, i])), 3) for f, i in zip(FEATS, clr_i)}, "conc": {f: round(float(np.median(D[m, i])), 3) for f, i in zip(FEATS, con_i)}, "size": round(float(np.median(D[m, NAMES.index("size")])), 3)}) # ---- forward-волатильність (конверт відкривається ТІЛЬКИ тут) ---- rf = A["rng_fwd"] base = float(np.median(rf)) fw = [] ps = [] for k in range(K): m = lab == k u = mannwhitneyu(rf[m], rf[~m], alternative="two-sided") ps.append(u.pvalue) fw.append({"cls": k, "med": round(float(np.median(rf[m])), 4), "mean": round(float(rf[m].mean()), 4), "ratio": round(float(np.median(rf[m]) / max(base, 1e-9)), 3), "p": float(u.pvalue)}) order = np.argsort(ps) q = np.empty(K) for rank, i in enumerate(order): q[i] = min(1.0, ps[i] * K / (rank + 1)) for i in range(K): fw[i]["q"] = float(round(q[i], 6)) out["fwd"] = {"base_med": round(base, 4), "cls": fw, "kruskal_p": float(kruskal(*[rf[lab == k] for k in range(K)]).pvalue)} # ---- чи не є якісь два класи однією купкою, розрізаною навпіл ---- day_ = ((ts - ts[0]) // 86400).astype(int) medd = np.array([np.median(rf[day_ == q]) if (day_ == q).sum() > 10 else np.nan for q in range(day_.max() + 1)]) rel_ = rf / medd[day_] Ck = np.array([Z[lab == k].mean(0) for k in range(K)]) rad = np.array([np.sqrt(((Z[lab == k] - Ck[k]) ** 2).sum(1).mean()) for k in range(K)]) pairs = [] for i in range(K): for j in range(i + 1, K): dist = float(np.linalg.norm(Ck[i] - Ck[j])); r = float((rad[i] + rad[j]) / 2) u1 = mannwhitneyu(rf[lab == i], rf[lab == j]) a = rel_[lab == i]; b = rel_[lab == j] u2 = mannwhitneyu(a[np.isfinite(a)], b[np.isfinite(b)]) pairs.append({"i": i, "j": j, "dist": round(dist, 3), "rad": round(r, 3), "ratio": round(dist / max(r, 1e-9), 3), "p_range": float(u1.pvalue), "p_within_day": float(u2.pvalue)}) flow_i = [NAMES.index(f"lev:{f}") for f in FEATS if f not in ("const_resist", "const_support")] fl_ = np.array([float(np.median(D[lab == k][:, flow_i].sum(axis=1))) for k in range(K)]) calm = [int(x) for x in np.argsort(fl_)[:2]] mm = np.isin(lab, calm) out["merge"] = {"pairs": pairs, "radius": [round(float(v), 3) for v in rad], "calm_cls": calm, "calm": {"share": round(float(mm.mean()), 4), "ratio": round(float(np.median(rf[mm]) / np.median(rf)), 3), "within_day": round(float(np.nanmedian(rel_[mm])), 3), "sil": round(float(silhouette(Z[mm], (lab[mm] == calm[0]).astype(int))), 4)}} # ---- конструктивна альтернатива: сітка терцилів на валідованих осях ---- # Якщо дані — континуум, чесна «класифікація» це не вигадані групи, а # ДЕТЕРМІНОВАНЕ розбиття по порогах на осях, порахованих на TRAIN. qa = np.quantile(Z[train, 0], [1 / 3, 2 / 3]) qb = np.quantile(Z[train, 1], [1 / 3, 2 / 3]) ga = np.digitize(Z[:, 0], qa); gb = np.digitize(Z[:, 1], qb) cell = ga * 3 + gb grid_cells = [] pg = [] for c in range(9): m = cell == c if m.sum() < 30: grid_cells.append({"cell": int(c), "n": int(m.sum())}); pg.append(1.0); continue u = mannwhitneyu(rf[m], rf[~m], alternative="two-sided") pg.append(u.pvalue) grid_cells.append({"cell": int(c), "a": int(c // 3), "b": int(c % 3), "n": int(m.sum()), "share": round(float(m.mean()), 4), "med": round(float(np.median(rf[m])), 4), "ratio": round(float(np.median(rf[m]) / max(base, 1e-9)), 3), "p": float(u.pvalue)}) og = np.argsort(pg); qg = np.empty(9) for rank, i in enumerate(og): qg[i] = min(1.0, pg[i] * 9 / (rank + 1)) for i in range(9): grid_cells[i]["q"] = float(round(qg[i], 6)) # відтворюваність правила: та сама клітинка при перебудові на іншій половині h = len(train) // 2 cells_h = [] for part in (train[:h], train[h:]): s2 = Space(repr_cols(rp), d).fit(D, part) Z2 = s2.transform(D) # знак осі довільний — вирівнюємо за кореляцією з основною for ax in (0, 1): if np.corrcoef(Z2[:, ax], Z[:, ax])[0, 1] < 0: Z2[:, ax] *= -1 q1 = np.quantile(Z2[part, 0], [1 / 3, 2 / 3]); q2 = np.quantile(Z2[part, 1], [1 / 3, 2 / 3]) cells_h.append(np.digitize(Z2[:, 0], q1) * 3 + np.digitize(Z2[:, 1], q2)) agree = float((cells_h[0][val] == cells_h[1][val]).mean()) out["grid"] = {"cells": grid_cells, "agree_halves": round(agree, 4), "kruskal_p": float(kruskal(*[rf[cell == c] for c in range(9) if (cell == c).sum() > 30]).pvalue)} # ---- чи корисні самі відстані (kNN, без моделі) ---- out["knn"] = {} for kk_ in (50, 200): out["knn"][str(kk_)] = knn_utility(Z, train, val, rf, k=kk_) try: Dn0 = null_desc(coin, seed=777) spn0 = Space(repr_cols(rp), d).fit(Dn0, train) Zn0 = spn0.transform(Dn0) out["knn"]["null200"] = knn_utility(Zn0, train, val, rf, k=200) del Dn0, Zn0 except Exception as e: out["knn"]["null200"] = {"err": str(e)[:100]} # ---- календарний конфаунд: чи не є «клас» просто шматком історії ---- import datetime as dt days = ((ts - ts[0]) // 86400).astype(int) month = np.array([dt.datetime.utcfromtimestamp(int(t)).month for t in ts]) hourw = np.array([(dt.datetime.utcfromtimestamp(int(t)).weekday() * 24 + dt.datetime.utcfromtimestamp(int(t)).hour) for t in ts]) v_m, p_m = cramers_v(lab, month) v_h, p_h = cramers_v(lab, hourw) half = ts < np.median(ts) out["calendar"] = { "cramers_v_month": round(v_m, 4), "p_month": p_m, "cramers_v_hourweek": round(v_h, 4), "p_hourweek": p_h, "share_first_half": [round(float((lab[half] == k).mean()), 4) for k in range(K)], "share_second_half": [round(float((lab[~half] == k).mean()), 4) for k in range(K)], "by_day": {str(k): [int(((lab == k) & (days == dd)).sum()) for dd in range(days.max() + 1)] for k in range(K)}} # ---- збіг з класами /events (той самий датасет, інший конвеєр) ---- out["vs_events"] = None try: ev = json.load(open(os.path.join(ROOT, "media", "analyst", "events", f"{coin}.json"))) st = json.load(open(os.path.join(ROOT, "media", "analyst", "standard", f"{coin}.json"))) el = np.array(ev["labels"]); et = np.array([e["t0"] for e in st["events"]]) if len(el) != len(et): # реєстри різної довжини — зіставляти мітки за порядком НЕ можна raise ValueError(f"реєстри не збігаються: {len(el)} проти {len(et)}") keep = el >= 0 wi = np.searchsorted(ts, et[keep]) - 1 ok = (wi >= 0) & (wi < len(ts)) from sklearn.metrics import adjusted_rand_score out["vs_events"] = {"n": int(ok.sum()), "ari": round(float(adjusted_rand_score(el[keep][ok], lab[wi[ok]])), 4)} except Exception as e: out["vs_events"] = {"err": str(e)[:120]} # ---- ряд для вкладки: мітка кожного вікна на осі часу ---- out["series"] = { "t0": int(ts[0]), "step": WIN, "k": [int(v) for v in ((ts - ts[0]) // WIN)], # позиція вікна в кроках "lab": [int(v) for v in lab], "px": [round(float(v), 1) for v in A["px"]], "fwd": [round(float(v), 3) for v in rf]} # ============================ графіки ============================ FG = "#e6edf3"; BG = "#0d1117"; GRID = "#30363d" plt.rcParams.update({"figure.facecolor": BG, "axes.facecolor": BG, "savefig.facecolor": BG, "text.color": FG, "axes.labelcolor": FG, "xtick.color": FG, "ytick.color": FG, "axes.edgecolor": GRID, "font.size": 9}) C1, C2, C3 = "#58a6ff", "#f85149", "#3fb950" g, kstar = gap_curve(mj, rp, d) ks = [x["K"] for x in g] fig, ax = plt.subplots(1, 2, figsize=(11, 4)) ax[0].plot(ks, [x["logw"] for x in g], "o-", color=C1, label="реальні дані") ax[0].plot(ks, [x["null_mean"] for x in g], "s--", color=C2, label="нуль-контроль") ax[0].set_xlabel("кількість класів K"); ax[0].set_ylabel("log W (розкид усередині класів)") ax[0].set_title(f"{coin}: чи стискає розбиття дані краще за шум · {rp}/d{d}") ax[0].legend(); ax[0].grid(alpha=.2) ax[1].errorbar(ks, [x["gap"] for x in g], yerr=[x["sk"] for x in g], fmt="o-", color=C3) ax[1].axhline(0, color=GRID) ax[1].set_xlabel("K"); ax[1].set_ylabel("gap = log W(нуль) − log W(дані)") ax[1].set_title("Gap-статистика: >0 = є структура понад шум") ax[1].grid(alpha=.2) fig.tight_layout(); fig.savefig(os.path.join(OUT_DIR, f"{coin}_gap.png"), dpi=110) plt.close(fig) real = next(s for s in mj["spaces"] if s["repr"] == rp and s["d"] == d) nulls = [next(s for s in rep if s["repr"] == rp and s["d"] == d) for rep in mj["null"]] fig, ax = plt.subplots(1, 2, figsize=(11, 4)) kk = [k for k in ks if k > 1] ax[0].plot(kk, [real["sil"][str(k)] for k in kk], "o-", color=C1, label="дані") ax[0].plot(kk, [np.mean([n["sil"][str(k)] for n in nulls]) for k in kk], "s--", color=C2, label="нуль-контроль") ax[0].set_xlabel("K"); ax[0].set_ylabel("силует"); ax[0].legend(); ax[0].grid(alpha=.2) ax[0].set_title("Силует: наскільки вікно ближче до свого класу, ніж до чужого") gm = real.get("gmm", {}).get("full", {}) if gm: icl = [gm[str(k)]["icl"] for k in ks if "icl" in gm[str(k)]] vll = [gm[str(k)]["val_ll"] for k in ks if "val_ll" in gm[str(k)]] a2 = ax[1]; a2.plot(ks[:len(icl)], icl, "o-", color=C1, label="ICL (менше — краще)") a2.set_xlabel("K"); a2.set_ylabel("ICL"); a2.grid(alpha=.2) a3 = a2.twinx(); a3.plot(ks[:len(vll)], vll, "s--", color=C3, label="правдоподібність на відкладених") a3.set_ylabel("logL / вікно (валідація)", color=C3) a2.set_title("Суміш гаусіан: вибір K") a2.legend(loc="lower left"); a3.legend(loc="lower right") fig.tight_layout(); fig.savefig(os.path.join(OUT_DIR, f"{coin}_select.png"), dpi=110) plt.close(fig) fig, ax = plt.subplots(1, 2, figsize=(11, 4.4)) ax[0].hexbin(Z[:, 0], Z[:, 1], gridsize=70, cmap="magma", bins="log") ax[0].set_title(f"{coin}: вікна в просторі описів (вісь 1 × вісь 2)") ax[0].set_xlabel("вісь 1"); ax[0].set_ylabel("вісь 2") Dn = null_desc(coin, seed=999) spn = Space(repr_cols(rp), d).fit(Dn, train); Zn = spn.transform(Dn) ax[1].hexbin(Zn[:, 0], Zn[:, 1], gridsize=70, cmap="magma", bins="log") ax[1].set_title("нуль-контроль (клітинки перемішано між вікнами)") ax[1].set_xlabel("вісь 1") del Dn, Zn fig.tight_layout(); fig.savefig(os.path.join(OUT_DIR, f"{coin}_map.png"), dpi=110) plt.close(fig) fig, ax = plt.subplots(1, 2, figsize=(12, 4.4)) M = np.array([[out["portrait"][k]["lev"][f] for f in FEATS] for k in range(K)]) im = ax[0].imshow(np.log10(np.maximum(M, 1e-3)), cmap="RdYlBu_r", aspect="auto") ax[0].set_xticks(range(8)); ax[0].set_xticklabels(FEATS, rotation=45, ha="right") ax[0].set_yticks(range(K)) ax[0].set_yticklabels([f"клас {k} ({out['shares'][k]*100:.0f}%)" for k in range(K)]) ax[0].set_title("Портрети класів: медіанний рівень фічі (log10, у частках свого масштабу)") fig.colorbar(im, ax=ax[0]) r = [x["ratio"] for x in fw] ax[1].bar(range(K), r, color=[C3 if x >= 1 else C2 for x in r]) ax[1].axhline(1, color=FG, lw=.8) ax[1].set_xticks(range(K)); ax[1].set_xticklabels([f"клас {k}" for k in range(K)]) ax[1].set_ylabel("розмах ціни наступної години / типовий") ax[1].set_title("Конверт: чи розрізняють класи майбутню волатильність") for i, x in enumerate(r): ax[1].text(i, x, f"{x:.2f}\nq={fw[i]['q']:.1g}", ha="center", va="bottom", fontsize=8) fig.tight_layout(); fig.savefig(os.path.join(OUT_DIR, f"{coin}_classes.png"), dpi=110) plt.close(fig) fig, ax = plt.subplots(figsize=(11, 3.6)) bd = np.array([out["calendar"]["by_day"][str(k)] for k in range(K)], float) bd = bd / np.maximum(bd.sum(0), 1) ax.stackplot(range(bd.shape[1]), bd, labels=[f"клас {k}" for k in range(K)]) ax.set_xlabel("день історії"); ax.set_ylabel("частка вікон") ax.set_title(f"{coin}: чи не є «клас» просто шматком історії · " f"V(міс)={v_m:.2f} V(год тижня)={v_h:.2f}") ax.legend(loc="upper right", ncol=K, fontsize=8) fig.tight_layout(); fig.savefig(os.path.join(OUT_DIR, f"{coin}_calendar.png"), dpi=110) plt.close(fig) with open(os.path.join(OUT_DIR, f"{coin}.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: звіт збережено → {OUT_DIR}") return out def ladder(coin, rp="full", d=16, n=3): """Драбина нулів: реальні дані → мʼякий нуль (форма фіч ціла, склад зруйновано) → жорсткий нуль (ціле зруйновано). Показує, ЩО саме тримає структуру.""" A = load_all(coin) D, train, val = A["D"], A["train"], A["val"] out = {"coin": coin, "repr": rp, "d": d, "ks": KS, "real": evaluate(D, train, val, rp, d, with_gmm=False), "series": [], "cells": []} log(f"{coin}: реальні sil(2)={out['real']['sil'][2]}") for b in range(n): Ds = null_series(coin, seed=5000 + b) out["series"].append(evaluate(Ds, train, val, rp, d, with_gmm=False, seed=SEED + b)) del Ds log(f" мʼякий нуль {b+1}/{n}: sil(2)={out['series'][-1]['sil'][2]}") Dc = null_desc(coin, seed=6000 + b) out["cells"].append(evaluate(Dc, train, val, rp, d, with_gmm=False, seed=SEED + b)) del Dc log(f" жорсткий нуль {b+1}/{n}: sil(2)={out['cells'][-1]['sil'][2]}") with open(os.path.join(CACHE, f"{coin}_ladder.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: драбина нулів збережена") return out # ============ чи корисні відстані: kNN-передбачення без жодної моделі ============ def knn_utility(Z, train, val, y, k=200, seed=SEED): """Суто відстаневий метод: прогноз = медіана forward-розмаху k найближчих сусідів з TRAIN. Оцінка — на відкладених вікнах (VAL). Якщо простір описів хоч щось знає про майбутнє, це видно тут, і жодна модель не потрібна.""" from sklearn.neighbors import NearestNeighbors from scipy.stats import spearmanr nn = NearestNeighbors(n_neighbors=k).fit(Z[train]) _, ind = nn.kneighbors(Z[val]) pred = np.median(y[train][ind], axis=1) act = y[val] rho = float(spearmanr(pred, act).statistic) o = np.argsort(pred) dec = np.array_split(o, 10) lift = [round(float(np.median(act[d]) / max(np.median(act), 1e-9)), 3) for d in dec] return {"k": k, "rho": round(rho, 4), "decile_lift": lift, "top_vs_bot": round(lift[-1] / max(lift[0], 1e-9), 3)} # ============ контроль стиснення: ті самі критерії на СИРОМУ векторі 1680 ============ def raw_matrix(coin, seed=None): """Вікно як вектор 1680 = log1p(x/масштаб) без жодних описів. Якщо seed заданий — той самий нуль-контроль (перемішування клітинок між вікнами).""" d = np.load(os.path.join(CACHE, f"{coin}_data.npz")) X = d["X"]; train = np.asarray(d["train"]); val = np.asarray(d["val"]) if seed is not None: rng = np.random.default_rng(seed) for part in (train, val): Xp = X[part] for j in range(8): for t in range(WIN): Xp[:, j, t] = Xp[rng.permutation(len(part)), j, t] X[part] = Xp del Xp s = np.load(os.path.join(CACHE, f"{coin}_feat.npz"))["scale"] Y = np.log1p(X / s[None, :, None]).reshape(len(X), -1) return Y, train, val def eval_raw(Y, train, val, d, ks=KS, seed=SEED): from sklearn.decomposition import PCA T = Y[train] med = np.median(T, axis=0); sd = T.std(axis=0); sd[sd < 1e-9] = 1.0 pca = PCA(n_components=d, whiten=True, svd_solver="randomized", random_state=SEED).fit((T - med) / sd) Ztr = pca.transform((T - med) / sd).astype(np.float32) res = {"repr": "raw", "d": d, "evr": round(float(pca.explained_variance_ratio_.sum()), 4), "nn": nn_stats(Ztr, seed=seed)} kw = kmeans_logw(Ztr, ks, seed=seed) res["logw"] = {K: round(kw[K][0], 5) for K in ks} res["sil"] = {K: (round(silhouette(Ztr, kw[K][1].labels_, seed=seed), 4) if K > 1 else None) for K in ks} return res def raws(coin, d=16, n_null=3): Y, train, val = raw_matrix(coin) log(f"{coin}: сирий вектор {Y.shape}") out = {"coin": coin, "ks": KS, "d": d, "real": eval_raw(Y, train, val, d)} log(f" реальні: evr={out['real']['evr']:.3f} sil(2)={out['real']['sil'][2]} " f"pd_cv={out['real']['nn']['pd_cv']:.3f}") del Y out["null"] = [] for b in range(n_null): Yn, _, _ = raw_matrix(coin, seed=3000 + b) out["null"].append(eval_raw(Yn, train, val, d, seed=SEED + b)) log(f" нуль {b+1}/{n_null}: sil(2)={out['null'][-1]['sil'][2]}") del Yn with open(os.path.join(CACHE, f"{coin}_raw.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: сирий контроль збережено") return out def confound(coin, rp="full", d=16, K=4): """Найважливіша перевірка при забороні зовнішньої норми: чи не є «клас» просто ЕПОХОЮ історії. Все міряється ще раз ВСЕРЕДИНІ ДОБИ — forward-розмах ділиться на медіану ТОГО Ж ДНЯ. Якщо різниця класів зникає — конвеєр класифікував календар, а не ринок.""" from sklearn.cluster import KMeans from scipy.stats import spearmanr, mannwhitneyu A = load_all(coin) D, train, val, ts = A["D"], A["train"], A["val"], A["ts"] rf = A["rng_fwd"] day = ((ts - ts[0]) // 86400).astype(int) med_d = np.zeros(day.max() + 1) for dd in range(day.max() + 1): m = day == dd med_d[dd] = np.median(rf[m]) if m.sum() > 10 else np.nan rel = rf / np.where(np.isfinite(med_d[day]) & (med_d[day] > 0), med_d[day], np.nan) sp = Space(repr_cols(rp), d).fit(D, train) Z = sp.transform(D) lab = KMeans(n_clusters=K, n_init=10, random_state=SEED).fit(Z[train]).predict(Z) out = {"coin": coin, "repr": rp, "d": d, "K": K, "cls": [], "grid": []} for k in range(K): m = (lab == k) & np.isfinite(rel) o = np.isfinite(rel) & (lab != k) out["cls"].append({"cls": k, "n": int(m.sum()), "abs_ratio": round(float(np.median(rf[lab == k]) / np.median(rf)), 3), "within_day": round(float(np.median(rel[m])), 3), "p": float(mannwhitneyu(rel[m], rel[o]).pvalue)}) qa = np.quantile(Z[train, 0], [1 / 3, 2 / 3]); qb = np.quantile(Z[train, 1], [1 / 3, 2 / 3]) cell = np.digitize(Z[:, 0], qa) * 3 + np.digitize(Z[:, 1], qb) for c in range(9): m = (cell == c) & np.isfinite(rel) if m.sum() < 30: continue out["grid"].append({"a": c // 3, "b": c % 3, "n": int(m.sum()), "abs_ratio": round(float(np.median(rf[cell == c]) / np.median(rf)), 3), "within_day": round(float(np.median(rel[m])), 3)}) # kNN: те саме, але з прибраним денним рівнем в обох рядах from sklearn.neighbors import NearestNeighbors nn = NearestNeighbors(n_neighbors=200).fit(Z[train]) _, ind = nn.kneighbors(Z[val]) pred = np.median(rf[train][ind], axis=1) ok = np.isfinite(rel[val]) out["knn"] = { "rho_raw": round(float(spearmanr(pred[ok], rf[val][ok]).statistic), 4), "rho_within_day": round(float(spearmanr( pred[ok] / med_d[day[val]][ok], rel[val][ok]).statistic), 4)} # скільки взагалі знає сам денний рівень (стеля «епохи») out["day_level_rho"] = round(float(spearmanr( med_d[day[val]][ok], rf[val][ok]).statistic), 4) with open(os.path.join(CACHE, f"{coin}_confound.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: конфаунд-тест — kNN ρ {out['knn']['rho_raw']} → " f"{out['knn']['rho_within_day']} усередині доби") return out def glevel(Xc, s): """Рівень фічі у вікні в разах від масштабу — ГЕОМЕТРИЧНИЙ (через log1p). Звичайне середнє тут не працює: його перекручує один тик.""" return np.expm1(np.log1p(Xc / s[None, :, None]).mean(axis=2)) def live(coin, rp="full", d=16, K=4, n_rand=40000, seed=SEED): """ЯК УПІЗНАТИ «ШТОРМ» У ПОТОЦІ. 1. Чесне число OOS: показники штормового класу лише на ВАЛІДАЦІЇ. 2. Дистиляція у ПРАВИЛО на 8 числах (геометричні рівні фіч у вікні) — щоб у живому режимі не тягнути PCA+kmeans і не залежати від сітки. Пороги підбираються на TRAIN, звіт — на VAL. 3. Правило прикладається до ВИПАДКОВИХ вікон з довільним зсувом (не кратним 210 с): якщо покриття збігається з сітковим, це стан ринку, а не артефакт нарізки. 4. Дрейф: як часто правило спрацьовує по місяцях (константи масштабу заморожені, а активність ринку їздить). """ from sklearn.cluster import KMeans A = load_all(coin) D, train, val, ts = A["D"], A["train"], A["val"], A["ts"] rf = A["rng_fwd"] day = ((ts - ts[0]) // 86400).astype(int) med_d = np.array([np.median(rf[day == dd]) if (day == dd).sum() > 10 else np.nan for dd in range(day.max() + 1)]) rel = rf / med_d[day] sp = Space(repr_cols(rp), d).fit(D, train) Z = sp.transform(D) lab = KMeans(n_clusters=K, n_init=10, random_state=SEED).fit(Z[train]).predict(Z) flow = [f for f in FEATS if f not in ("const_resist", "const_support")] fl = np.array([float(np.median(D[lab == k][:, [NAMES.index(f"lev:{f}") for f in flow]].sum(axis=1))) for k in range(K)]) st = int(np.argmax(fl)) out = {"coin": coin, "storm_cls": st, "n_rand": n_rand} # --- 1. лише валідація --- def block(idx, m): return {"n": int(m.sum()), "share": round(float(m.mean()), 4), "ratio": round(float(np.median(rf[idx][m]) / np.median(rf[idx])), 3), "within_day": round(float(np.nanmedian(rel[idx][m])), 3)} out["oos"] = {"val": block(val, lab[val] == st), "train": block(train, lab[train] == st), "all": block(np.arange(len(lab)), lab == st)} # --- 2. правило на 8 числах --- # у нашому просторі описів lev = mean(log1p(x/s)); повертаємо в «рази від масштабу» L = np.expm1(D[:, [NAMES.index(f"lev:{f}") for f in FEATS]]) i_dr = [FEATS.index("resist_minus"), FEATS.index("support_minus")] i_in = [FEATS.index("resist_plus"), FEATS.index("support_plus")] i_wl = [FEATS.index("const_resist"), FEATS.index("const_support")] drain = np.sqrt(L[:, i_dr[0]] * L[:, i_dr[1]]) infl = np.sqrt(L[:, i_in[0]] * L[:, i_in[1]]) wall = np.sqrt(L[:, i_wl[0]] * L[:, i_wl[1]]) y = (lab == st) best = None for a in np.quantile(drain[train], np.arange(0.60, 0.99, 0.01)): for b in np.quantile(infl[train], np.arange(0.50, 0.98, 0.02)): for c in np.quantile(wall[train], np.arange(0.30, 0.95, 0.05)): p = (drain[train] > a) & (infl[train] > b) & (wall[train] < c) tp = float((p & y[train]).sum()) if tp < 100: continue prec = tp / max(p.sum(), 1); rec = tp / max(y[train].sum(), 1) f1 = 2 * prec * rec / max(prec + rec, 1e-9) if best is None or f1 > best["f1"]: best = {"drain_gt": float(a), "infl_gt": float(b), "wall_lt": float(c), "f1": float(f1), "prec": float(prec), "rec": float(rec)} rule = (drain > best["drain_gt"]) & (infl > best["infl_gt"]) & (wall < best["wall_lt"]) def qual(idx): p = rule[idx]; t = y[idx] tp = float((p & t).sum()) return {"share": round(float(p.mean()), 4), "prec": round(tp / max(p.sum(), 1), 3), "rec": round(tp / max(t.sum(), 1), 3), "ratio": round(float(np.median(rf[idx][p]) / np.median(rf[idx])), 3), "within_day": round(float(np.nanmedian(rel[idx][p])), 3)} out["rule"] = {k: round(v, 4) for k, v in best.items()} out["rule_train"] = qual(train) out["rule_val"] = qual(val) # --- 4. дрейф спрацювань по місяцях --- import datetime as dt mon = np.array([dt.datetime.utcfromtimestamp(int(t)).strftime("%Y-%m") for t in ts]) out["by_month"] = {m: round(float(rule[mon == m].mean()), 4) for m in sorted(set(mon.tolist()))} # --- 3. випадкові вікна з довільним зсувом --- log(f"{coin}: перечитую ряд для вікон з довільним зсувом…") S = ST.load(coin) _, first = np.unique(S["ts"], return_index=True) first = np.sort(first) for k in ["ts", "price"] + FEATS: S[k] = S[k][first] t0 = int(S["ts"][0]); t1 = int(S["ts"][-1]); N = t1 - t0 + 1 pos = (S["ts"] - t0).astype(np.int64) present = np.zeros(N, bool); present[pos] = True cp = np.concatenate([[0], np.cumsum(present, dtype=np.int64)]) gi = np.arange(N, dtype=np.float64) px = np.interp(gi, pos, S["price"]).astype(np.float32) rng = np.random.default_rng(seed) cand = rng.integers(0, N - WIN - 3600, size=n_rand * 2) ok = (cp[cand + WIN] - cp[cand]) / WIN >= MIN_REAL starts = cand[ok][:n_rand] idx = starts[:, None] + np.arange(WIN, dtype=np.int64)[None, :] s = np.load(os.path.join(CACHE, f"{coin}_feat.npz"))["scale"] G = np.empty((len(starts), 8), np.float32) for j, f in enumerate(FEATS): v = np.interp(gi, pos, S[f].astype(np.float64)).astype(np.float32) G[:, j] = np.expm1(np.log1p(v[idx] / s[j]).mean(axis=1)) del v dr = np.sqrt(G[:, i_dr[0]] * G[:, i_dr[1]]) inf_ = np.sqrt(G[:, i_in[0]] * G[:, i_in[1]]) wl = np.sqrt(G[:, i_wl[0]] * G[:, i_wl[1]]) hit = (dr > best["drain_gt"]) & (inf_ > best["infl_gt"]) & (wl < best["wall_lt"]) fr = np.empty(len(starts), np.float32) for i, a0 in enumerate(starts): seg = px[a0 + WIN:a0 + WIN + 3600] fr[i] = (seg.max() - seg.min()) / max(px[a0], 1e-9) * 100 if len(seg) else 0.0 dayr = ((t0 + starts) - t0) // 86400 medr = np.array([np.median(fr[dayr == q]) if (dayr == q).sum() > 10 else np.nan for q in range(int(dayr.max()) + 1)]) relr = fr / medr[dayr] out["random"] = { "n": int(len(starts)), "share": round(float(hit.mean()), 4), "base_range": round(float(np.median(fr)), 4), "range": round(float(np.median(fr[hit])), 4), "ratio": round(float(np.median(fr[hit]) / np.median(fr)), 3), "within_day": round(float(np.nanmedian(relr[hit])), 3), "offset_mod210": int(np.median(starts % WIN))} with open(os.path.join(CACHE, f"{coin}_live.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: правило drain>{best['drain_gt']:.3f} · infl>{best['infl_gt']:.3f} · " f"wall<{best['wall_lt']:.3f} → VAL точність {out['rule_val']['prec']} " f"повнота {out['rule_val']['rec']} · випадкові вікна {out['random']['share']*100:.1f}% " f"часу, розмах ×{out['random']['ratio']}") return out def learn(coin, rp="full", d=16, K=4, seed=SEED): """ЧИ ВАРТО ВЧИТИ МОДЕЛЬ НА «ШТОРМ». Три задачі на одних і тих самих описах вікна, щоб побачити різницю: A. відтворити МІТКУ класу. Мітка — детермінована функція цих самих описів (kmeans на PCA від них), тож модель зобовʼязана її вивчити майже точно. Високий AUC тут не означає нічого про ринок: це тавтологія. B. передбачити розмах наступної години (абсолютний). Тут модель має право вивчити ЕПОХУ: у високоактивні місяці все підряд волатильніше. C. передбачити розмах ВІДНОСНО МЕДІАНИ ТОГО Ж ДНЯ. Епоха викреслена, лишається те, що вікно справді знає. Це чесна задача. Скрізь: вчимо на TRAIN (з ембарго), міряємо на VAL. Плюс нуль-контроль для C і порівняння з правилом на 3 порогах при ОДНАКОВІЙ частоті спрацювань. """ from sklearn.ensemble import HistGradientBoostingClassifier as HGC from sklearn.ensemble import HistGradientBoostingRegressor as HGR from sklearn.metrics import roc_auc_score, f1_score from sklearn.cluster import KMeans from scipy.stats import spearmanr A = load_all(coin) D, train, val, ts = A["D"], A["train"], A["val"], A["ts"] rf = A["rng_fwd"] day = ((ts - ts[0]) // 86400).astype(int) med_d = np.array([np.median(rf[day == dd]) if (day == dd).sum() > 10 else np.nan for dd in range(day.max() + 1)]) rel = rf / med_d[day] sp = Space(repr_cols(rp), d).fit(D, train) Z = sp.transform(D) lab = KMeans(n_clusters=K, n_init=10, random_state=SEED).fit(Z[train]).predict(Z) flow = [f for f in FEATS if f not in ("const_resist", "const_support")] fl = [float(np.median(D[lab == k][:, [NAMES.index(f"lev:{f}") for f in flow]].sum(1))) for k in range(K)] stc = int(np.argmax(fl)) out = {"coin": coin, "storm_cls": stc} # ---- A. відтворити мітку ---- yA = (lab == stc).astype(int) mA = HGC(max_iter=300, learning_rate=0.08, random_state=seed).fit(D[train], yA[train]) pA = mA.predict_proba(D[val])[:, 1] out["A_label"] = {"auc": round(float(roc_auc_score(yA[val], pA)), 4), "f1": round(float(f1_score(yA[val], (pA > 0.5).astype(int))), 4), "share": round(float(yA[val].mean()), 4)} # ---- B. абсолютний розмах ---- yB = np.log1p(rf) mB = HGR(max_iter=400, learning_rate=0.06, random_state=seed).fit(D[train], yB[train]) pB = mB.predict(D[val]) out["B_abs"] = {"rho": round(float(spearmanr(pB, rf[val]).statistic), 4)} # ---- C. розмах відносно того ж дня ---- okt = np.isfinite(rel[train]); okv = np.isfinite(rel[val]) yC = np.log(np.maximum(rel, 1e-6)) mC = HGR(max_iter=400, learning_rate=0.06, random_state=seed).fit( D[train][okt], yC[train][okt]) pC = mC.predict(D[val]) out["C_rel"] = {"rho": round(float(spearmanr(pC[okv], rel[val][okv]).statistic), 4)} # ---- порівняння при ОДНАКОВІЙ частоті спрацювань ---- V = None p = os.path.join(CACHE, f"{coin}_live.json") if os.path.exists(p): V = json.load(open(p)) rate = V["rule_val"]["share"] if V else 0.093 def at_rate(score, name): thr = np.quantile(score, 1 - rate) m = score >= thr return {"name": name, "rate": round(float(m.mean()), 4), "ratio": round(float(np.median(rf[val][m]) / np.median(rf[val])), 3), "within_day": round(float(np.nanmedian(rel[val][m])), 3)} cmpr = [at_rate(pB, "модель B (абсолютний розмах)"), at_rate(pC, "модель C (усередині доби)"), at_rate(pA, "модель A (мітка класу)")] if V: cmpr.append({"name": "правило на 3 порогах", "rate": V["rule_val"]["share"], "ratio": V["rule_val"]["ratio"], "within_day": V["rule_val"]["within_day"]}) out["at_rate"] = cmpr # ---- нуль-контроль для C ---- Dn = null_desc(coin, seed=4242) mN = HGR(max_iter=400, learning_rate=0.06, random_state=seed).fit( Dn[train][okt], yC[train][okt]) pN = mN.predict(Dn[val]) out["C_null"] = {"rho": round(float(spearmanr(pN[okv], rel[val][okv]).statistic), 4)} del Dn # ---- на чому модель C тримається ---- from sklearn.inspection import permutation_importance sub = np.random.default_rng(seed).choice(np.flatnonzero(okv), min(2500, okv.sum()), replace=False) pi = permutation_importance(mC, D[val][sub], yC[val][sub], n_repeats=3, random_state=seed, n_jobs=1) o = np.argsort(-pi.importances_mean)[:12] out["importance"] = [[NAMES[i], round(float(pi.importances_mean[i]), 4)] for i in o] with open(os.path.join(CACHE, f"{coin}_learn.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: A(мітка) AUC {out['A_label']['auc']} · B(абс) ρ {out['B_abs']['rho']} · " f"C(усередині доби) ρ {out['C_rel']['rho']} при нулі {out['C_null']['rho']}") return out # ============ КЛАСИФІКАЦІЯ ПОДІЙ «ШТОРМ» на повній матриці 8×210 ============ def storm_labels(coin, rp="full", d=16, K=4, target="storm"): """Ціль навчання. Визначається ТІЛЬКИ з боку входу (рівень потоків), ніколи за майбутньою ціною: storm — клас із найвищим сумарним рівнем 6 потокових фіч; calm — ОБʼЄДНАННЯ двох класів з найнижчим рівнем (їхні портрети й поведінка ціни збігаються, k-середні розрізали одну купку навпіл); mid — клас, що лишився (третій за рівнем потоків) — «клас 2» у нумерації звіту: стіни тонкі, потоки звичайні. Рахується один раз і заморожується — далі це просто y.""" from sklearn.cluster import KMeans A = load_all(coin) D, train = A["D"], A["train"] sp = Space(repr_cols(rp), d).fit(D, train) Z = sp.transform(D) lab = KMeans(n_clusters=K, n_init=10, random_state=SEED).fit(Z[train]).predict(Z) flow = [f for f in FEATS if f not in ("const_resist", "const_support")] fl = [float(np.median(D[lab == k][:, [NAMES.index(f"lev:{f}") for f in flow]].sum(1))) for k in range(K)] order = list(np.argsort(fl)) if target == "calm": pick = [int(order[0]), int(order[1])] return np.isin(lab, pick).astype(np.int8), pick if target == "mid": k2 = int(order[2]) return (lab == k2).astype(np.int8), k2 st = int(np.argmax(fl)) return (lab == st).astype(np.int8), st def pca_poles(coin, norm_days=28, sub_n=15000, mcs=100, seed=SEED, with_null=True): """ПЕРЕРАХУНОК полюсів «активний»/«тихий» ЗА СТАНДАРТОМ /states, але на НАШИХ вікнах і на ПОТОЧНОМУ датасеті. Чужий файл media/analyst/states не читаємо: мітка мусить оновлюватись разом з даними, як усе інше в цій лінії. Стандарт /states дослівно: фіча ділиться на власну ПРИЧИННУ норму за 28 діб → обрізка [1e-3, 1e4] → логарифм → робастний масштаб по TRAIN → PCA 32 → UMAP(10, сусідів 30) → HDBSCAN(min_cluster_size=100). Полюси називаються за рівнем активності, а не за номером, який дав HDBSCAN. Різниця з оригіналом одна й навмисна: вікна беруться наші (стик у стик, крок 210 с), а не їхні кожні 300 с — щоб мітка лягала рівно на той самий набір вікон, на якому потім вчиться модель. """ from sklearn.decomposition import PCA from sklearn.cluster import HDBSCAN os.makedirs(CACHE, exist_ok=True) dd = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") wts = np.asarray(dd["ts"]); train = np.asarray(dd["train"]) log(f"{coin}: перечитую ряд для норми {norm_days} діб…") S = ST.load(coin) _, first = np.unique(S["ts"], return_index=True) first = np.sort(first) for k in ["ts", "price"] + FEATS: S[k] = S[k][first] ts = S["ts"]; t0 = int(ts[0]); t1 = int(ts[-1]); N = t1 - t0 + 1 pos = (ts - t0).astype(np.int64) present = np.zeros(N, bool); present[pos] = True gi = np.arange(N, dtype=np.float64) NW = norm_days * 86400 starts = (wts - t0).astype(np.int64) keep = starts >= NW # прогрів норми геть starts = starts[keep] idx = starts[:, None] + np.arange(WIN, dtype=np.int64)[None, :] A = np.empty((len(starts), 8, WIN), np.float32) cnt = np.concatenate([[0], np.cumsum(present, dtype=np.int64)]) nreal = (cnt[NW + 1:] - cnt[1:N - NW + 1]).astype(np.float64) for jf, f in enumerate(FEATS): v = np.interp(gi, pos, S[f].astype(np.float64)) c = np.concatenate([[0.0], np.cumsum(v * present)]) nv = np.full(N, np.nan) with np.errstate(invalid="ignore", divide="ignore"): nv[NW:] = (c[NW + 1:] - c[1:N - NW + 1]) / np.where(nreal > 0, nreal, np.nan) r = (v / np.where(nv > 1e-12, nv, np.nan)).astype(np.float32) A[:, jf, :] = r[idx] del v, c, nv, r del gi, idx ok = np.isfinite(A).all(axis=(1, 2)) A = A[ok]; starts = starts[ok] wid = np.flatnonzero(keep)[ok] # номери НАШИХ вікон log(f"{coin}: вікон з нормою {len(A)} з {len(wts)} (прогрів {norm_days} діб геть)") tr_mask = np.isin(wid, train) tr_local = np.flatnonzero(tr_mask) F = np.clip(A, 1e-3, 1e4).astype(np.float32) act = np.log(F).mean(axis=(1, 2)) # рівень активності вікна (для назв) np.log(F, out=F) v = F[tr_local].reshape(-1) med = float(np.median(v)); iqr = float(np.subtract(*np.percentile(v, [75, 25]))) or 1.0 F = ((F - med) / iqr).reshape(len(F), -1) del A, v rng = np.random.default_rng(seed) sub = np.sort(rng.choice(tr_local, size=min(sub_n, len(tr_local)), replace=False)) def pipeline(FF): import umap p = PCA(n_components=32, random_state=seed).fit(FF[sub]) Z = p.transform(FF).astype(np.float32) u = umap.UMAP(n_components=10, n_neighbors=30, min_dist=0.0, metric="euclidean", random_state=seed, verbose=False).fit(Z[sub]) Zu = u.transform(Z).astype(np.float32) lb = HDBSCAN(min_cluster_size=int(mcs), cluster_selection_method="eom").fit_predict(Zu) return lb, float(p.explained_variance_ratio_.sum()) t0c = time.time() lab, evr = pipeline(F) ks = sorted({int(c) for c in lab if c >= 0}) log(f"{coin}: HDBSCAN дав {len(ks)} класів, шуму {float((lab<0).mean())*100:.1f}% " f"(evr {evr:.3f}, {time.time()-t0c:.0f}с)") out = {"coin": coin, "built": int(time.time()), "norm_days": norm_days, "mcs": mcs, "sub_n": int(len(sub)), "t0": int(wts[0]), "t1": int(wts[-1]), "days": round((wts[-1] - wts[0]) / 86400, 1), "n_win_all": int(len(wts)), "n_win": int(len(F)), "evr": round(evr, 4), "k": len(ks), "noise": round(float((lab < 0).mean()), 4), "classes": []} if not ks: log("класів немає"); return out amed = {c: float(np.median(act[lab == c])) for c in ks} hi = max(amed, key=amed.get); lo = min(amed, key=amed.get) out["hi_id"] = int(hi); out["lo_id"] = int(lo) # нуль-контроль: ті самі клітинки, перемішані між вікнами if with_null: Fn = F.reshape(len(F), 8, WIN).copy() for jf in range(8): for tt in range(WIN): Fn[:, jf, tt] = Fn[rng.permutation(len(Fn)), jf, tt] ln, _ = pipeline(Fn.reshape(len(Fn), -1)) out["null_k"] = len({int(c) for c in ln if c >= 0}) out["null_noise"] = round(float((ln < 0).mean()), 4) log(f"{coin}: нуль-контроль дав {out['null_k']} класів, шуму " f"{out['null_noise']*100:.1f}%") del Fn, ln # портрети і поведінка ціни (конверт) Aall = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") rf = np.asarray(Aall["rng_fwd"]); tsw = np.asarray(Aall["ts"]) day = ((tsw - tsw[0]) // 86400).astype(int) medd = np.array([np.median(rf[day == q]) if (day == q).sum() > 10 else np.nan for q in range(day.max() + 1)]) rel = rf / medd[day] base = float(np.median(rf)) G = np.expm1(np.log1p(np.asarray(Aall["X"])[wid] / np.load(os.path.join(CACHE, f"{coin}_feat.npz"))["scale"][None, :, None] ).mean(axis=2)) for c in ks: m = lab == c nm = "активний" if c == hi else ("тихий" if c == lo else f"проміжний {c}") out["classes"].append({ "id": int(c), "name": nm, "n": int(m.sum()), "share": round(float(m.mean()), 4), "share_all": round(float(m.sum() / len(wts)), 4), "act": round(float(amed[c]), 3), "lvl": {f: round(float(np.median(G[m, i])), 4) for i, f in enumerate(FEATS)}, "ratio": round(float(np.median(rf[wid][m]) / base), 3), "within_day": round(float(np.nanmedian(rel[wid][m])), 3)}) out["classes"].sort(key=lambda x: -x["act"]) # мітка на ВСІ наші вікна: -2 = поза прогрівом норми full = np.full(len(wts), -2, np.int16) full[wid] = lab np.savez(os.path.join(CACHE, f"{coin}_pcapoles.npz"), lab=full, wid=wid, act=act) with open(os.path.join(CACHE, f"{coin}_pcapoles.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}: полюси перераховано · активний {out['classes'][0]['share_all']*100:.1f}% " f"· тихий {out['classes'][-1]['share_all']*100:.1f}% від усіх вікон") return out def states_labels(coin, target): """Ціль з лінії /states: полюс одного з движків, зіставлений з нашими вікнами за часом. Повертає (y, idx) — мітки і номери НАШИХ вікон, для яких мітка є. Назви: pcaquiet — «тихе ПСА» (полюс «тихий» движка PCA), pcaactive, aequiet, aeactive. Мітка береться як є; навчання далі йде за стандартом /wclass — вхід 8×210 сирих значень, перетворення лише всередині вікна. """ eng = "pca" if target.startswith("pca") else "ae" pole = "quiet" if target.endswith("quiet") else "active" # ГОЛОВНИЙ ШЛЯХ: наш власний перерахунок полюсів на поточному датасеті own = os.path.join(CACHE, f"{coin}_pcapoles.json") if eng == "pca" and os.path.exists(own): meta = json.load(open(own)) lab = np.load(os.path.join(CACHE, f"{coin}_pcapoles.npz"))["lab"] tid = meta["lo_id"] if pole == "quiet" else meta["hi_id"] idx = np.flatnonzero(lab > -2) # вікна з нормою y = (lab[idx] == tid).astype(np.int8) return y, idx, {"engine": "pca(наш перерахунок)", "pole": pole, "cls_id": int(tid), "built": meta["built"], "norm_days": meta["norm_days"]} sp = os.path.join(ROOT, "media", "analyst", "states", f"{coin}.json") S = json.load(open(sp)) sts = np.array(S["series"]["ts"]); lb = np.array(S["series"][eng]) act = np.array(S["series"]["act"]) ids = [c["id"] for c in S["engines"][eng]["classes"] if c["id"] >= 0] med = {i: float(np.median(act[lb == i])) for i in ids if (lb == i).any()} tid = (min(med, key=med.get) if pole == "quiet" else max(med, key=med.get)) A = load_all(coin); ts = A["ts"] j = np.searchsorted(ts, sts) - 1 ok = (j >= 0) & (j < len(ts)) & (np.abs(ts[np.clip(j, 0, len(ts) - 1)] - sts) <= WIN) idx = j[ok] y = (lb[ok] == tid).astype(np.int8) # одне наше вікно могло зіставитись двічі — лишаємо перше _, first = np.unique(idx, return_index=True) first = np.sort(first) return y[first], idx[first], {"engine": eng, "pole": pole, "cls_id": int(tid)} PREP = os.environ.get("WCLASS_PREP", "scale") # ВАРІАНТИ ВХОДУ. Питання, яке вони вирішують: скільки коштує СТРОГЕ правило # «у вікно не входить жодне число ззовні». Варіант scale ставить коліно логарифма # там, де фіча реально живе (масштаб узятий з трейну — тобто статистика датасету). # Решта варіантів такої статистики не використовують взагалі: коліно призначене # наперед, і вікно перетворюється функцією ЛИШЕ від власних 1680 чисел. PREP_DESC = { "scale": "log1p(x/масштаб) + стандартизація по каналу — масштаб і центр із трейну", "pure": "log1p(x) — жодного числа ззовні вікна", "fixed": "log1p(x/0.01) — стала призначена наперед, не з даних", "asinh": "asinh(x) — жодного числа ззовні вікна", } def _prep_fold(X, tr, va, mode=None): """Перетворення ВСЕРЕДИНІ вікна. Усі константи (якщо вони взагалі є) — тільки з трейн-частини фолда, валідація на них не впливає.""" mode = mode or PREP if mode == "pure": return (np.log1p(X[tr]).astype(np.float32), np.log1p(X[va]).astype(np.float32)) if mode == "fixed": return (np.log1p(X[tr] / 0.01).astype(np.float32), np.log1p(X[va] / 0.01).astype(np.float32)) if mode == "asinh": return (np.arcsinh(X[tr]).astype(np.float32), np.arcsinh(X[va]).astype(np.float32)) s = X[tr].mean(axis=(0, 2)).astype(np.float32) s[s <= 0] = 1.0 Ltr = np.log1p(X[tr] / s[None, :, None]) m = Ltr.mean(axis=(0, 2), dtype=np.float64).astype(np.float32) sd = Ltr.std(axis=(0, 2), dtype=np.float64).astype(np.float32) sd[sd < 1e-6] = 1.0 Lva = (np.log1p(X[va] / s[None, :, None]) - m[None, :, None]) / sd[None, :, None] Ltr = (Ltr - m[None, :, None]) / sd[None, :, None] return Ltr.astype(np.float32), Lva.astype(np.float32) def _cnn(nch=8, T=WIN): import torch.nn as nn return nn.Sequential( nn.Conv1d(nch, 32, 7, padding=3), nn.BatchNorm1d(32), nn.ReLU(), nn.MaxPool1d(2), nn.Conv1d(32, 64, 5, padding=2), nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2), nn.Conv1d(64, 96, 3, padding=1), nn.BatchNorm1d(96), nn.ReLU(), nn.AdaptiveAvgPool1d(1), nn.Flatten(), nn.Dropout(0.2), nn.Linear(96, 1)) def _fit_cnn(Xtr, ytr, Xva, epochs=8, bs=512, lr=2e-3, seed=SEED): import torch, torch.nn as nn torch.manual_seed(seed) torch.set_num_threads(max(1, os.cpu_count() or 4)) m = _cnn(Xtr.shape[1], Xtr.shape[2]) pos = float(ytr.mean()) lossf = nn.BCEWithLogitsLoss(pos_weight=torch.tensor([(1 - pos) / max(pos, 1e-6)])) opt = torch.optim.AdamW(m.parameters(), lr=lr, weight_decay=1e-4) Xt = torch.from_numpy(Xtr); yt = torch.from_numpy(ytr.astype(np.float32)) n = len(Xt) for ep in range(epochs): m.train(); perm = torch.randperm(n) for i in range(0, n, bs): j = perm[i:i + bs] opt.zero_grad() out = m(Xt[j]).squeeze(1) loss = lossf(out, yt[j]) loss.backward(); opt.step() m.eval() ps = [] with torch.no_grad(): Xv = torch.from_numpy(Xva) for i in range(0, len(Xv), 2048): ps.append(torch.sigmoid(m(Xv[i:i + 2048]).squeeze(1)).numpy()) return np.concatenate(ps) def clf(coin, folds=10, models=("gbm", "cnn"), seed=SEED, target="storm"): """10 випадкових частин: у кожному прогоні 9 частин трейн / 1 валідація. Вхід усіх моделей — уся матриця 8×210. Кожне вікно дістає прогноз рівно раз.""" from sklearn.linear_model import LogisticRegression from sklearn.ensemble import HistGradientBoostingClassifier as HGC from sklearn.decomposition import PCA from sklearn.neighbors import KNeighborsClassifier from sklearn.metrics import roc_auc_score, average_precision_score, f1_score sub = None if target.startswith(("pca", "ae")): y, sub, st = states_labels(coin, target) else: y, st = storm_labels(coin, target=target) dd = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") X = np.asarray(dd["X"]) if sub is None else np.asarray(dd["X"])[sub] n = len(X) rng = np.random.default_rng(seed) part = rng.permutation(n) % folds # 10 випадкових частин log(f"{coin}: вхід «{PREP}» · {n} вікон · ціль «{target}» {int(y.sum())} ({y.mean()*100:.1f}%) · " f"{folds} частин по {n//folds}") oof = {k: np.full(n, np.nan) for k in models} times = {k: 0.0 for k in models} for f in range(folds): va = np.flatnonzero(part == f); tr = np.flatnonzero(part != f) t0 = time.time() Ltr, Lva = _prep_fold(X, tr, va) Ftr = Ltr.reshape(len(tr), -1); Fva = Lva.reshape(len(va), -1) log(f" фолд {f+1}/{folds}: трейн {len(tr)} · вал {len(va)} " f"(підготовка {time.time()-t0:.0f}с)") if "logreg" in models: t = time.time() mo = LogisticRegression(max_iter=300, C=0.3).fit(Ftr, y[tr]) oof["logreg"][va] = mo.predict_proba(Fva)[:, 1] times["logreg"] += time.time() - t if "gbm" in models: t = time.time() mo = HGC(max_iter=120, learning_rate=0.12, max_bins=32, early_stopping=True, validation_fraction=0.1, random_state=seed).fit(Ftr, y[tr]) oof["gbm"][va] = mo.predict_proba(Fva)[:, 1] times["gbm"] += time.time() - t if "knn" in models: t = time.time() p = PCA(n_components=64, whiten=True, svd_solver="randomized", random_state=seed).fit(Ftr) mo = KNeighborsClassifier(n_neighbors=25, n_jobs=-1).fit(p.transform(Ftr), y[tr]) oof["knn"][va] = mo.predict_proba(p.transform(Fva))[:, 1] times["knn"] += time.time() - t if "cnn" in models: t = time.time() oof["cnn"][va] = _fit_cnn(Ltr, y[tr], Lva, seed=seed + f) times["cnn"] += time.time() - t del Ltr, Lva, Ftr, Fva log(" " + " · ".join(f"{k} AUC {roc_auc_score(y[va], oof[k][va]):.4f}" for k in models if np.isfinite(oof[k][va]).all())) res = {"coin": coin, "target": target, "cls": (st if isinstance(st, (list, dict)) else int(st)), "folds": folds, "n": int(n), "pos": float(y.mean()), "models": {}} for k in models: p = oof[k] ok = np.isfinite(p) per = [] for f in range(folds): m = (part == f) & ok if m.sum() < 10 or len(np.unique(y[m])) < 2: continue per.append([float(roc_auc_score(y[m], p[m])), float(average_precision_score(y[m], p[m]))]) per = np.array(per) thr = np.quantile(p[ok], 1 - y.mean()) pred = (p >= thr).astype(int) tp = float(((pred == 1) & (y == 1) & ok).sum()) res["models"][k] = { "auc": round(float(per[:, 0].mean()), 4), "auc_sd": round(float(per[:, 0].std()), 4), "ap": round(float(per[:, 1].mean()), 4), "ap_sd": round(float(per[:, 1].std()), 4), "f1": round(float(f1_score(y[ok], pred[ok])), 4), "prec": round(tp / max(pred[ok].sum(), 1), 4), "rec": round(tp / max(float(y[ok].sum()), 1), 4), "sec": round(times[k], 1)} log(f"{k:7s} AUC {res['models'][k]['auc']:.4f}±{res['models'][k]['auc_sd']:.4f} · " f"PR-AUC {res['models'][k]['ap']:.4f} · F1 {res['models'][k]['f1']:.3f} · " f"точність {res['models'][k]['prec']:.3f} · повнота {res['models'][k]['rec']:.3f} " f"· {res['models'][k]['sec']:.0f}с") sfx = "" if target == "storm" else "_" + target if PREP != "scale": sfx = f"_{target}_{PREP}" res["prep"] = PREP res["prep_desc"] = PREP_DESC.get(PREP, PREP) np.savez(os.path.join(CACHE, f"{coin}_{'prep' if PREP!='scale' else 'clf'}_oof{sfx}.npz"), y=y, part=part, sub=(np.arange(n, dtype=np.int64) if sub is None else np.asarray(sub)), **{k: oof[k] for k in models}) res["best"] = max(res["models"], key=lambda k: res["models"][k]["ap"]) with open(os.path.join(CACHE, f"{coin}_{'prep' if PREP!='scale' else 'clf'}{sfx}.json"), "w") as fh: json.dump(res, fh, ensure_ascii=False) log(f"{coin}: краща модель — {res['best']}") return res # ============ ПОЛЮСИ /states: «активний» і «тихий» ============ def _fit_cnn_multi(Xtr, ytr, Xva, ncls=3, epochs=8, bs=512, lr=2e-3, seed=SEED): import torch, torch.nn as nn torch.manual_seed(seed) torch.set_num_threads(max(1, os.cpu_count() or 4)) m = _cnn(Xtr.shape[1], Xtr.shape[2]) m[-1] = nn.Linear(96, ncls) cnt = np.bincount(ytr, minlength=ncls).astype(np.float64) w = torch.tensor((cnt.sum() / np.maximum(cnt, 1)) / ncls, dtype=torch.float32) lossf = nn.CrossEntropyLoss(weight=w) opt = torch.optim.AdamW(m.parameters(), lr=lr, weight_decay=1e-4) Xt = torch.from_numpy(Xtr); yt = torch.from_numpy(ytr.astype(np.int64)) n = len(Xt) for ep in range(epochs): m.train(); perm = torch.randperm(n) for i in range(0, n, bs): k = perm[i:i + bs] opt.zero_grad() loss = lossf(m(Xt[k]), yt[k]) loss.backward(); opt.step() m.eval() ps = [] with torch.no_grad(): Xv = torch.from_numpy(Xva) for i in range(0, len(Xv), 2048): ps.append(torch.softmax(m(Xv[i:i + 2048]), dim=1).numpy()) return np.concatenate(ps) MODELS_DIR = os.path.join(OUT_DIR, "models") HOLD_FRAC = 0.02 # скільки датасету відкладаємо для перевірки продакшен-моделі HOLD_CHUNKS = 4 # і скількома суцільними шматками (0.5% кожен) HOLD_EMB = 9 # ембарго навколо шматка, вікон (~30 хв) — сусіди корелюють 0.76 def holdout_blocks(n, frac=HOLD_FRAC, chunks=HOLD_CHUNKS, emb=HOLD_EMB, seed=SEED): """Відкладена вибірка: `chunks` суцільних шматків у випадкових місцях ряду. Суцільних, а не розсипаних поодинці, бо сусідні вікна корелюють 0.76 — розсипана вибірка дала б завищену оцінку. Навколо кожного шматка ембарго.""" m = max(1, int(round(n * frac / chunks))) rng = np.random.default_rng(seed) hold = np.zeros(n, bool); ban = np.zeros(n, bool) placed = 0 for _ in range(2000): if placed >= chunks: break a = int(rng.integers(0, n - m)) b = a + m if ban[max(0, a - emb):min(n, b + emb)].any(): continue hold[a:b] = True ban[max(0, a - emb):min(n, b + emb)] = True placed += 1 train = np.flatnonzero(~ban) return np.flatnonzero(hold), train, placed def final(coin, targets=("storm", "calm", "mid", "pcaquiet"), seed=SEED, epochs=8): """ПРОДАКШЕН-МОДЕЛІ: переможець (Conv1D) вчиться на ВСІХ вікнах кожної цілі й лягає на диск разом з константами перетворення і порогом. Це те, що потім крутиться на живих даних — жодного підбору тут уже немає, конфігурація зафіксована крос-валідацією на етапі clf.""" import torch os.makedirs(MODELS_DIR, exist_ok=True) dd = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") Xm = dd["X"] # лишається на диску, в памʼять не тягнемо out = {} import gc for tg in targets: p = os.path.join(CACHE, f"{coin}_clf{'' if tg == 'storm' else '_' + tg}.json") if not os.path.exists(p): log(f"{coin}/{tg}: немає крос-валідації — пропускаю"); continue cv = json.load(open(p)) if tg.startswith(("pca", "ae")): y, sub, meta = states_labels(coin, tg) else: y, meta = storm_labels(coin, target=tg); sub = None n_all = len(y) hold, tr_idx, placed = holdout_blocks(n_all) log(f"{coin}/{tg}: відкладено {len(hold)} вікон ({len(hold)/n_all*100:.1f}%) " f"{placed} шматками · навчання на {len(tr_idx)}") # константи (лише для варіанта scale) — рахуються на навчальній частині s = m = sd = None if PREP == "scale": acc = np.zeros(8, np.float64); cnt = 0 for i in range(0, len(tr_idx), 4000): ii = tr_idx[i:i + 4000] if sub is None else sub[tr_idx[i:i + 4000]] ch = np.asarray(Xm[ii]); acc += ch.mean(axis=(0, 2)) * len(ii); cnt += len(ii) s = (acc / cnt).astype(np.float32); s[s <= 0] = 1.0 # перетворення шматками у заздалегідь виділений масив L = np.empty((n_all, 8, WIN), np.float32) for i in range(0, n_all, 4000): ii = np.arange(i, min(i + 4000, n_all)) ch = np.asarray(Xm[ii if sub is None else sub[ii]]) if PREP == "pure": L[ii] = np.log1p(ch) elif PREP == "fixed": L[ii] = np.log1p(ch / 0.01) elif PREP == "asinh": L[ii] = np.arcsinh(ch) else: L[ii] = np.log1p(ch / s[None, :, None]) del ch if PREP == "scale": m = L[tr_idx].mean(axis=(0, 2), dtype=np.float64).astype(np.float32) sd = L[tr_idx].std(axis=(0, 2), dtype=np.float64).astype(np.float32) sd[sd < 1e-6] = 1.0 L -= m[None, :, None]; L /= sd[None, :, None] gc.collect() t0 = time.time() import torch.nn as nn torch.manual_seed(seed); torch.set_num_threads(max(1, os.cpu_count() or 4)) net = _cnn(8, WIN) pos = float(y[tr_idx].mean()) lossf = nn.BCEWithLogitsLoss(pos_weight=torch.tensor([(1 - pos) / max(pos, 1e-6)])) opt = torch.optim.AdamW(net.parameters(), lr=2e-3, weight_decay=1e-4) Xa = torch.from_numpy(L) # вид на той самий масив, без копії yt = torch.from_numpy(y.astype(np.float32)) ti = torch.from_numpy(tr_idx.astype(np.int64)) for ep in range(epochs): net.train(); perm = ti[torch.randperm(len(ti))] for i in range(0, len(perm), 512): k = perm[i:i + 512] opt.zero_grad() loss = lossf(net(Xa[k]).squeeze(1), yt[k]); loss.backward(); opt.step() net.eval() with torch.no_grad(): ps = np.concatenate([torch.sigmoid(net(Xa[i:i + 2048]).squeeze(1)).numpy() for i in range(0, n_all, 2048)]) thr = float(np.quantile(ps[tr_idx], 1 - float(y[tr_idx].mean()))) torch.save(net.state_dict(), os.path.join(MODELS_DIR, f"{coin}_{tg}.pt")) log(f"{coin}/{tg}: згортка навчена · {time.time()-t0:.0f}с") del opt, Xa, yt, ti; gc.collect() # ---- бустинг на тому самому вході ---- import pickle from sklearn.ensemble import HistGradientBoostingClassifier as HGC F = L.reshape(n_all, -1) tg0 = time.time() gb = HGC(max_iter=120, learning_rate=0.12, max_bins=32, early_stopping=True, validation_fraction=0.1, random_state=seed).fit(F[tr_idx], y[tr_idx]) pg = np.concatenate([gb.predict_proba(F[i:i + 8192])[:, 1] for i in range(0, n_all, 8192)]) thr_g = float(np.quantile(pg[tr_idx], 1 - float(y[tr_idx].mean()))) with open(os.path.join(MODELS_DIR, f"{coin}_{tg}_gbm.pkl"), "wb") as fh: pickle.dump(gb, fh) log(f"{coin}/{tg}: бустинг навчений · {time.time()-tg0:.0f}с") from sklearn.metrics import roc_auc_score, average_precision_score def hmet(pp, t): yy = y[hold]; sc = pp[hold]; prd = sc >= t tp = float((prd & (yy == 1)).sum()) return {"auc": round(float(roc_auc_score(yy, sc)), 4), "ap": round(float(average_precision_score(yy, sc)), 4), "prec": round(tp / max(prd.sum(), 1), 4), "rec": round(tp / max(float(yy.sum()), 1), 4), "fires": round(float(prd.mean()), 4)} hold_cnn = hmet(ps, thr); hold_gbm = hmet(pg, thr_g) log(f"{coin}/{tg}: на відкладених 2% — згортка AUC {hold_cnn['auc']} PR {hold_cnn['ap']} · " f"бустинг AUC {hold_gbm['auc']} PR {hold_gbm['ap']}") meta_out = {"coin": coin, "target": tg, "built": int(time.time()), "prep": PREP, "prep_desc": PREP_DESC.get(PREP, PREP), "n": int(n_all), "n_train": int(len(tr_idx)), "n_hold": int(len(hold)), "hold_chunks": int(placed), "pos": round(float(y.mean()), 4), "scale": ([float(v) for v in s] if s is not None else None), "mean": ([float(v) for v in m] if m is not None else None), "std": ([float(v) for v in sd] if sd is not None else None), "thr": round(thr, 5), "thr_gbm": round(thr_g, 5), "epochs": epochs, "win": WIN, "hold_cnn": hold_cnn, "hold_gbm": hold_gbm, "cv_auc": cv["models"]["cnn"]["auc"], "cv_ap": cv["models"]["cnn"]["ap"], "cv_auc_gbm": cv["models"].get("gbm", {}).get("auc"), "cv_ap_gbm": cv["models"].get("gbm", {}).get("ap"), "label_meta": meta if isinstance(meta, dict) else {"cls": meta}} del L, F, ps, pg; gc.collect() with open(os.path.join(MODELS_DIR, f"{coin}_{tg}.json"), "w") as fh: json.dump(meta_out, fh, ensure_ascii=False) out[tg] = meta_out log(f"{coin}/{tg}: збережено · пороги {thr:.3f} / {thr_g:.3f}") return out def scan(coin, step=1, seed=SEED): """ПРОДАКШЕН-МОДЕЛІ НА ВСЬОМУ ДАТАСЕТІ. Ті самі файли, що крутяться наживо, прикладені до всіх вікон історії — щоб бачити поведінку моделі не на 10 годинах живого потоку, а на 171 добі. ЧЕСНО ПРО ЦЕЙ ГРАФІК: 98% вікон були в навчанні цих моделей, тож тут переважно памʼять, а не передбачення. Єдина чесна частина — 4 відкладені шматки по 0.5%, вони позначені окремо. Для чесної оцінки на всій історії дивись out-of-fold прогнози в розділі ⑤ (кожне вікно оцінене копією, яка його не бачила). """ import torch, pickle dd = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") Xm = dd["X"]; ts = np.asarray(dd["ts"]); px = np.asarray(dd["px"]) rf = np.asarray(dd["rng_fwd"]) n_all = len(ts) models = {} for p in sorted(__import__("glob").glob(os.path.join(MODELS_DIR, f"{coin}_*.json"))): meta = json.load(open(p)) wp = p[:-5] + ".pt" if not os.path.exists(wp): continue net = _cnn(8, WIN); net.load_state_dict(torch.load(wp, map_location="cpu")) net.eval() gp = os.path.join(MODELS_DIR, f"{coin}_{meta['target']}_gbm.pkl") gb = pickle.load(open(gp, "rb")) if os.path.exists(gp) else None models[meta["target"]] = {"net": net, "gbm": gb, "meta": meta} if not models: log("продакшен-моделей немає"); return None log(f"{coin}: {len(models)} цілей · вікон {n_all}") out = {tg: {"cnn": np.empty(n_all, np.float32), "gbm": (np.empty(n_all, np.float32) if models[tg]["gbm"] is not None else None)} for tg in models} t0 = time.time() for i in range(0, n_all, 4000): j = min(i + 4000, n_all) ch = np.asarray(Xm[i:j]) for tg, M in models.items(): mode = M["meta"].get("prep", "scale") if mode == "pure": L = np.log1p(ch).astype(np.float32) elif mode == "fixed": L = np.log1p(ch / 0.01).astype(np.float32) elif mode == "asinh": L = np.arcsinh(ch).astype(np.float32) else: s = np.array(M["meta"]["scale"], np.float32) m = np.array(M["meta"]["mean"], np.float32) sd = np.array(M["meta"]["std"], np.float32) L = ((np.log1p(ch / s[None, :, None]) - m[None, :, None]) / sd[None, :, None]).astype(np.float32) with torch.no_grad(): out[tg]["cnn"][i:j] = torch.sigmoid( M["net"](torch.from_numpy(L)).squeeze(1)).numpy() if M["gbm"] is not None: out[tg]["gbm"][i:j] = M["gbm"].predict_proba(L.reshape(len(L), -1))[:, 1] del L del ch if (i // 4000) % 5 == 0: log(f" {i}/{n_all}") log(f"{coin}: прогін завершено за {time.time()-t0:.0f}с") # відкладені шматки кожної цілі — позначаємо, бо лише вони чесні hold_marks = {} for tg in models: nt = models[tg]["meta"]["n"] h, _, _ = holdout_blocks(nt) if models[tg]["meta"].get("label_meta", {}).get("engine", "").startswith("pca"): _, sub, _ = states_labels(coin, tg) h = sub[h] hold_marks[tg] = [int(v) for v in h] res = {"coin": coin, "built": int(time.time()), "win": WIN, "n": int(n_all), "t0": int(ts[0]), "step": step, "k": [int(v) for v in (ts - ts[0]) // WIN][::step], "px": [round(float(v), 1) for v in px][::step], "fwd": [round(float(v), 3) for v in rf][::step], "targets": {}} for tg in models: M = models[tg]["meta"] res["targets"][tg] = { "thr": M["thr"], "thr_gbm": M.get("thr_gbm"), "pos": M["pos"], "prep": M.get("prep"), "built": M["built"], "hold_cnn": M.get("hold_cnn"), "hold_gbm": M.get("hold_gbm"), "n_hold": M.get("n_hold"), "hold_chunks": M.get("hold_chunks"), "cnn": [int(round(float(v) * 100)) for v in out[tg]["cnn"]][::step], "gbm": ([int(round(float(v) * 100)) for v in out[tg]["gbm"]][::step] if out[tg]["gbm"] is not None else None), "hold": hold_marks[tg]} p = os.path.join(OUT_DIR, "scan.json") with open(p, "w") as fh: json.dump(res, fh, ensure_ascii=False) log(f"{coin}: {p} · {os.path.getsize(p)/1e6:.1f} МБ") return res def poles(coin, engine="ae", folds=5, seed=SEED): """Відтворити два полюси лінії /states («активний» і «тихий») з самого вікна. Мітки беруться з media/analyst/states/{COIN}.json і зіставляються з нашими вікнами за часом. Далі три рівні складності: 1. ОДНЕ ЧИСЛО — геометричний рівень 6 потокових фіч у вікні (два пороги); 2. ті самі 89 описів вікна; 3. уся матриця 8×210. Спліт випадковий на folds частин, кожне вікно оцінене рівно раз. """ from sklearn.linear_model import LogisticRegression from sklearn.ensemble import HistGradientBoostingClassifier as HGC from sklearn.metrics import roc_auc_score, average_precision_score sp = os.path.join(ROOT, "media", "analyst", "states", f"{coin}.json") if not os.path.exists(sp): log("немає media/analyst/states — запусти states_report.py"); return None S = json.load(open(sp)) sts = np.array(S["series"]["ts"]); lab_s = np.array(S["series"][engine]) act = np.array(S["series"]["act"]) ids = [c["id"] for c in S["engines"][engine]["classes"] if c["id"] >= 0] med = {i: float(np.median(act[lab_s == i])) for i in ids if (lab_s == i).any()} hi = max(med, key=med.get); lo = min(med, key=med.get) A = load_all(coin); ts = A["ts"]; D = A["D"] dd = np.load(os.path.join(CACHE, f"{coin}_data.npz"), mmap_mode="r") X = np.asarray(dd["X"]) s = np.load(os.path.join(CACHE, f"{coin}_feat.npz"))["scale"] j = np.searchsorted(ts, sts) - 1 okm = (j >= 0) & (j < len(ts)) & (np.abs(ts[np.clip(j, 0, len(ts) - 1)] - sts) <= WIN) idx = j[okm] y = np.zeros(len(idx), np.int64) y[lab_s[okm] == hi] = 1 # активний y[lab_s[okm] == lo] = 2 # тихий log(f"{coin}/{engine}: зіставлено {len(idx)} вікон · активних {int((y==1).sum())} " f"({(y==1).mean()*100:.1f}%) · тихих {int((y==2).sum())} ({(y==2).mean()*100:.1f}%)") # ---- рівень 1: одне число ---- FL = [i for i, f in enumerate(FEATS) if f not in ("const_resist", "const_support")] G = np.empty(len(idx), np.float32) for i in range(0, len(idx), 4000): ch = np.asarray(X[idx[i:i + 4000]]) L = np.log1p(ch / s[None, :, None]).mean(axis=2) G[i:i + 4000] = np.expm1(L[:, FL].mean(axis=1)) def best_thr(yy, sc, up): qs = np.quantile(sc, np.arange(0.002, 1.0, 0.002)); best = None for t in qs: p = (sc > t) if up else (sc < t) tp = float((p & (yy == 1)).sum()) if tp < 10: continue pr = tp / max(p.sum(), 1); rc = tp / max(yy.sum(), 1) f1 = 2 * pr * rc / max(pr + rc, 1e-9) if best is None or f1 > best["f1"]: best = {"f1": round(f1, 4), "thr": round(float(t), 4), "prec": round(pr, 4), "rec": round(rc, 4), "share": round(float(p.mean()), 4)} return best out = {"coin": coin, "engine": engine, "folds": folds, "n": int(len(idx)), "share_act": round(float((y == 1).mean()), 4), "share_quiet": round(float((y == 2).mean()), 4), "rule": {"active": best_thr((y == 1).astype(int), G, True), "quiet": best_thr((y == 2).astype(int), G, False)}, "level": {"act_med": round(float(np.median(G[y == 1])), 4), "quiet_med": round(float(np.median(G[y == 2])), 4), "other_med": round(float(np.median(G[y == 0])), 4)}} log(f" правило на одному числі: активний F1 {out['rule']['active']['f1']} · " f"тихий F1 {out['rule']['quiet']['f1']}") # ---- рівні 2 і 3: моделі ---- rng = np.random.default_rng(seed) part = rng.permutation(len(idx)) % folds P = {k: np.full((len(idx), 3), np.nan) for k in ("desc", "gbm", "cnn")} for f in range(folds): va = np.flatnonzero(part == f); tr = np.flatnonzero(part != f) Ltr, Lva = _prep_fold(X, idx[tr], idx[va]) Ftr = Ltr.reshape(len(tr), -1); Fva = Lva.reshape(len(va), -1) # 2: наші 89 описів Dtr = D[idx[tr]]; Dva = D[idx[va]] mo = HGC(max_iter=200, learning_rate=0.1, random_state=seed).fit(Dtr, y[tr]) P["desc"][va] = mo.predict_proba(Dva) # 3a: бустинг на 1680 mo = HGC(max_iter=120, learning_rate=0.12, max_bins=32, early_stopping=True, validation_fraction=0.1, random_state=seed).fit(Ftr, y[tr]) P["gbm"][va] = mo.predict_proba(Fva) # 3b: згортка на 8×210 P["cnn"][va] = _fit_cnn_multi(Ltr, y[tr], Lva, seed=seed + f) del Ltr, Lva, Ftr, Fva log(f" фолд {f+1}/{folds} готовий") out["models"] = {} for k, pr in P.items(): r = {} for cid, nm in ((1, "active"), (2, "quiet")): yy = (y == cid).astype(int); sc = pr[:, cid] thr = np.quantile(sc, 1 - yy.mean()) p = sc >= thr tp = float((p & (yy == 1)).sum()) r[nm] = {"auc": round(float(roc_auc_score(yy, sc)), 4), "ap": round(float(average_precision_score(yy, sc)), 4), "prec": round(tp / max(p.sum(), 1), 4), "rec": round(tp / max(float(yy.sum()), 1), 4)} out["models"][k] = r log(f" {k}: активний AUC {r['active']['auc']} PR {r['active']['ap']} · " f"тихий AUC {r['quiet']['auc']} PR {r['quiet']['ap']}") np.savez(os.path.join(CACHE, f"{coin}_poles_{engine}.npz"), y=y, part=part, idx=idx, G=G, **{k: v for k, v in P.items()}) with open(os.path.join(CACHE, f"{coin}_poles_{engine}.json"), "w") as fh: json.dump(out, fh, ensure_ascii=False) log(f"{coin}/{engine}: полюси збережено") return out if __name__ == "__main__": cmd = sys.argv[1] if len(sys.argv) > 1 else "data" coin = sys.argv[2] if len(sys.argv) > 2 else "BTC" if cmd == "data": build_data(coin) elif cmd == "feat": build_feat(coin) elif cmd == "model": model(coin) elif cmd == "tests": tests(coin) elif cmd == "raws": raws(coin) elif cmd == "ladder": ladder(coin) elif cmd == "confound": confound(coin) elif cmd == "live": live(coin) elif cmd == "learn": learn(coin) elif cmd == "final": final(coin, tuple(sys.argv[3].split(",")) if len(sys.argv) > 3 else ("storm", "calm", "mid", "pcaquiet")) elif cmd == "scan": scan(coin) elif cmd == "pcapoles": pca_poles(coin) elif cmd == "poles": poles(coin, engine=sys.argv[3] if len(sys.argv) > 3 else "ae", folds=int(sys.argv[4]) if len(sys.argv) > 4 else 5) elif cmd == "clf": clf(coin, folds=int(sys.argv[3]) if len(sys.argv)>3 else 10, target=sys.argv[4] if len(sys.argv)>4 else "storm") elif cmd == "report": rp = sys.argv[3] if len(sys.argv) > 3 else "full" d = int(sys.argv[4]) if len(sys.argv) > 4 else 16 K = int(sys.argv[5]) if len(sys.argv) > 5 else 4 report(coin, rp, d, K) else: log(f"невідомий етап: {cmd}") ``` ### Живий контур `wclass_live.py` — читає свіжі секунди з датасету, застосовує продакшен-моделі, тримає 48 год історії ```python """ ЖИВА ПЕРЕВІРКА МОДЕЛЕЙ /wclass на даних колектора. Що робить: бере СВІЖІ секунди з Desktop/DATA (той самий read-only майстер-датасет), ріже їх на вікна по 210 с точно так само, як етап `data`, проганяє через збережені продакшен-моделі (`media/analyst/wclass/models/{COIN}_{ціль}.pt`) і дописує результат у `media/analyst/wclass/live.json`. Навіщо окремо від офлайн-звіту: у звіті прогноз out-of-fold — чесний, але порахований на історії, яку модель бачила в сусідніх фолдах. Тут модель зустрічає дані, яких не існувало на момент навчання. Це єдина перевірка, яку не можна підкрутити. Стандарти ті самі, що в лінії: · вікно 210 с, вікна стик у стик, приймається при ≥95% справжніх секунд; · перетворення лише всередині вікна: log1p(x / масштаб) → стандартизація по каналу, константи взяті з файлу моделі (пораховані на навчанні, не на живих даних); · поріг спрацювання теж з файлу моделі. Ціна пишеться поруч, але в модель не входить — вона лише для графіка і для перевірки постфактум (розмах наступної години зʼявляється із запізненням на годину). Запуск: python3 wclass_live.py [COIN] (launchd: com.at.wclasslive, крок 210 с = вікно) """ import os, sys, json, time, glob import numpy as np import pandas as pd import config from standard_signals import FEATS ROOT = os.path.dirname(os.path.abspath(__file__)) OUT_DIR = os.path.join(ROOT, "media", "analyst", "wclass") MODELS_DIR = os.path.join(OUT_DIR, "models") LIVE = os.path.join(OUT_DIR, "live.json") WIN = 210 MIN_REAL = 0.95 KEEP_H = 48 # скільки годин історії тримаємо у файлі LOOKBACK_H = 6 # скільки годин перечитуємо з датасету за прогін COLS = ["timestamp", "price_avg"] + FEATS def log(*a): print(time.strftime("[%H:%M:%S]"), *a, flush=True) def read_tail(coin, hours=LOOKBACK_H): """Останні N годин монети з майстер-датасету (лише читання).""" files = config.find_files(coin) if not files: return None frames = [] for _, path in files[-2:]: try: frames.append(pd.read_csv(path, usecols=COLS)) except Exception: continue if not frames: return None d = pd.concat(frames, ignore_index=True) d = d.drop_duplicates(subset="timestamp").sort_values("timestamp") tmax = int(d["timestamp"].iloc[-1]) d = d[d["timestamp"] >= tmax - hours * 3600] return d, tmax def cut_windows(d): """Суцільна секундна шкала → вікна стик у стик, як в етапі data. ВАЖЛИВО: старти прибиті до ГЛОБАЛЬНОЇ сітки (t кратне 210 с від епохи Unix), а не до початку прочитаного шматка. Інакше кожен прогін ріже вікна з нового місця, вони перекриваються, і в файл лягає 17 копій того самого часу замість одного ряду. Саме так і сталося на першому запуску.""" ts = d["timestamp"].values.astype(np.int64) t0, t1 = int(ts[0]), int(ts[-1]) N = t1 - t0 + 1 pos = (ts - t0) present = np.zeros(N, bool); present[pos] = True gi = np.arange(N, dtype=np.float64) px = np.interp(gi, pos, d["price_avg"].values.astype(np.float64)).astype(np.float32) first = (-t0) % WIN # зсув до найближчого кратного 210 с starts = np.arange(first, N - WIN + 1, WIN, dtype=np.int64) cp = np.concatenate([[0], np.cumsum(present, dtype=np.int64)]) real = (cp[starts + WIN] - cp[starts]) / WIN starts = starts[real >= MIN_REAL] if not len(starts): return None idx = starts[:, None] + np.arange(WIN, dtype=np.int64)[None, :] X = np.empty((len(starts), 8, WIN), np.float32) for j, f in enumerate(FEATS): v = np.interp(gi, pos, d[f].values.astype(np.float64)).astype(np.float32) X[:, j, :] = v[idx] return X, (t0 + starts).astype(np.int64), px[starts], px[np.minimum(starts + WIN, N - 1)] def load_models(coin): """Обидва продакшен-движки на кожну ціль: згортка (.pt) і бустинг (_gbm.pkl).""" import torch, pickle import wclass as W out = {} for p in sorted(glob.glob(os.path.join(MODELS_DIR, f"{coin}_*.json"))): meta = json.load(open(p)) wp = p[:-5] + ".pt" if not os.path.exists(wp): continue net = W._cnn(8, WIN) net.load_state_dict(torch.load(wp, map_location="cpu")) net.eval() gb = None gp = os.path.join(MODELS_DIR, f"{coin}_{meta['target']}_gbm.pkl") if os.path.exists(gp): with open(gp, "rb") as fh: gb = pickle.load(fh) out[meta["target"]] = {"cnn": net, "gbm": gb, "meta": meta} return out def prep_input(X, meta): """Те саме перетворення, що й на навчанні. Який саме варіант — записано в файлі моделі: `scale` тягне 24 константи, решта не тягне нічого.""" mode = meta.get("prep", "scale") if mode == "pure": return np.log1p(X).astype(np.float32) if mode == "fixed": return np.log1p(X / 0.01).astype(np.float32) if mode == "asinh": return np.arcsinh(X).astype(np.float32) s = np.array(meta["scale"], np.float32) m = np.array(meta["mean"], np.float32) sd = np.array(meta["std"], np.float32) return ((np.log1p(X / s[None, :, None]) - m[None, :, None]) / sd[None, :, None]).astype(np.float32) def apply_models(models, X): import torch res = {} for tg, M in models.items(): meta = M["meta"] L = prep_input(X, meta) with torch.no_grad(): p = torch.sigmoid(M["cnn"](torch.from_numpy(L)).squeeze(1)).numpy() r = {"p": p, "thr": float(meta["thr"])} if M["gbm"] is not None: r["pg"] = M["gbm"].predict_proba(L.reshape(len(L), -1))[:, 1] r["thr_gbm"] = float(meta.get("thr_gbm", meta["thr"])) res[tg] = r return res def run(coin="BTC"): os.makedirs(OUT_DIR, exist_ok=True) models = load_models(coin) if not models: log(f"{coin}: моделей немає — запусти python3 wclass.py final {coin}") return None got = read_tail(coin) if got is None: log(f"{coin}: даних немає"); return None d, tmax = got lag = int(time.time()) - tmax cut = cut_windows(d) if cut is None: log(f"{coin}: жодного повного вікна"); return None X, wts, px0, px1 = cut pr = apply_models(models, X) log(f"{coin}: вікон {len(X)} · лаг даних {lag} с · моделей {len(models)}") st = {"coin": coin, "win": WIN, "targets": {}, "rows": []} if os.path.exists(LIVE): try: st = json.load(open(LIVE)) except Exception: pass st["coin"] = coin; st["win"] = WIN st["built"] = int(time.time()); st["data_lag_s"] = lag st["targets"] = {tg: {"thr": pr[tg]["thr"], "thr_gbm": pr[tg].get("thr_gbm"), "has_gbm": "pg" in pr[tg], "prep": models[tg]["meta"].get("prep", "scale"), "cv_auc": models[tg]["meta"].get("cv_auc"), "cv_ap": models[tg]["meta"].get("cv_ap"), "cv_auc_gbm": models[tg]["meta"].get("cv_auc_gbm"), "hold_cnn": models[tg]["meta"].get("hold_cnn"), "hold_gbm": models[tg]["meta"].get("hold_gbm"), "pos": models[tg]["meta"].get("pos"), "built": models[tg]["meta"].get("built")} for tg in pr} have = {int(r["t"]) for r in st.get("rows", [])} for i, t in enumerate(wts): if int(t) in have: continue row = {"t": int(t), "px": round(float(px0[i]), 2), "px_end": round(float(px1[i]), 2), **{tg: int(round(float(pr[tg]["p"][i]) * 100)) for tg in pr}} for tg in pr: if "pg" in pr[tg]: row[tg + "_g"] = int(round(float(pr[tg]["pg"][i]) * 100)) st["rows"].append(row) st["rows"] = sorted(st["rows"], key=lambda r: r["t"])[-int(KEEP_H * 3600 / WIN):] # розмах наступної години — дописуємо заднім числом, коли година вже минула ts_all = np.array([r["t"] for r in st["rows"]]) px_all = np.array([r["px"] for r in st["rows"]], np.float64) for k, r in enumerate(st["rows"]): if "fwd" in r: continue a = r["t"] + WIN; b = a + 3600 m = (ts_all >= a) & (ts_all <= b) if m.sum() >= 15 and ts_all[m].max() >= b - WIN * 2: seg = px_all[m] r["fwd"] = round(float((seg.max() - seg.min()) / max(r["px"], 1e-9) * 100), 3) with open(LIVE, "w") as fh: json.dump(st, fh, ensure_ascii=False) fired = {tg: sum(1 for r in st["rows"][-20:] if r.get(tg, 0) / 100 >= pr[tg]["thr"]) for tg in pr} log(f"{coin}: у файлі {len(st['rows'])} вікон · за останню годину спрацювань " + " · ".join(f"{tg} {fired[tg]}/20" for tg in fired)) return st if __name__ == "__main__": run(sys.argv[1] if len(sys.argv) > 1 else "BTC") ``` ### Вкладка дашборда `web/templates/wclass.html` — 13 розділів; сторож test_pages.js перевіряє її на реальних даних ```html Класифікація вікон 210 с · market_dashboard
← хаб /events /ae /standard

🧩 Класифікація вікон 210 с — строго за їхніми 1680 числами

завантаження…

① Дані та обмеження

Вікна стик у стик, вхід моделі — рівно 8 фіч × 210 секунд СИРИХ значень. Жодної багатодобової норми, ціни, часу чи сусідніх вікон: ззовні беруться лише константи масштабу, пораховані на TRAIN. Спліт блоками з ембарго — бо сусідні вікна корелюють.

② Чи є класи взагалі

Кожен критерій має власний нуль-контроль: клітинки «фіча × секунда» перемішуються МІЖ вікнами (гине все, крім маргіналій) і весь конвеєр женеться заново. Якщо критерій на шумі дає те саме, що на даних, — він нічого не довів.
gap = наскільки розбиття стискає дані краще, ніж шум

③ Пастка: два поважні критерії, які тут брешуть

Бутстреп-LRT і ICL обирають K упевнено — і роблять це однаково на реальних даних і на чистому шумі. Вони міряють негаусовість маргіналій, а не наявність груп.

④ Де сидить структура насправді

Драбина нулів: мʼякий нуль перемішує між вікнами ЦІЛІ ряди фіч (форма кожної фічі всередині вікна ціла, спільність фіч зруйнована), жорсткий — окремі клітинки. Різниця показує, що саме тримає структуру.

⑤ Клас «шторм» на осі часу

Ціна за той самий період — для контексту. Обидва графіки масштабуються разом.
колесо = масштаб по часу · тягни = зсув · подвійний клік = увесь період
Нижче — частка вибраного класу в кожній колонці пікселів. Коли наближаєш достатньо, стовпчики стають окремими вікнами по 210 с.
колесо = масштаб · тягни = зсув · подвійний клік = скинути
А під ним — що на цьому ж часі кажуть НАВЧЕНІ моделі. Показано прогноз найкращої моделі кожної цілі, і не абиякий, а out-of-fold: кожне вікно оцінене тією копією моделі, яка його не бачила. Яскравість = впевненість.
спільний масштаб із графіками вище

⑥ Конфаунд епохи: те саме, але всередині доби

Без зовнішньої норми найсильніше в даних — рівень активності, який їздить від місяця до місяця. Тому кожне число переміряне ще раз усередині доби: forward-розмах поділено на медіану ТОГО Ж дня. Що не пережило поправку — то був календар, а не стан ринку.

⑦ Що лишається як робоча класифікація

Класи не знайдені — тож розбиття задається порогами: терцилі осі 1 (рівень) × терцилі осі 2 (нахил), пороги пораховані на TRAIN.

⑧ Як упізнати «шторм» у потоці

Живому боту не потрібні ні PCA, ні k-середні: клас дистильовано у три порівняння на шести геометричних рівнях (звичайне середнє тут не працює — його перекручує один тик). Пороги підібрані на TRAIN, усі числа нижче — на ВАЛІДАЦІЇ. Окремо правило прикладене до 40 тис. вікон з ДОВІЛЬНИМ зсувом (не кратним 210 с): якщо покриття збігається з сітковим, це стан ринку, а не артефакт нарізки.
Константи масштабу заморожені на TRAIN, а активність ринку їздить від місяця до місяця — тому частота спрацювань дрейфує. Це ціна умови «нічого ззовні вікна».
частка часу, коли правило спрацьовує

⑨ Чи допоможе навчена модель

Три різні задачі на одних і тих самих описах вікна. Відтворити МІТКУ класу — тавтологія: мітка сама є функцією цих описів, тож модель зобовʼязана її вивчити. Передбачити абсолютний розмах — задача, де модель має право вивчити ЕПОХУ. Чесна задача одна: розмах ВІДНОСНО МЕДІАНИ ТОГО Ж ДНЯ.
А тепер головне порівняння: усіх поставлено на ОДНАКОВУ частоту спрацювань (~9%) і поміряно на валідації.

⑩ Класифікація подій ШТОРМ — порівняння моделей

Вхід кожної моделі — уся матриця вікна 8 × 210 сирих значень, без наших описів. Перетворення лише всередині вікна: log1p(x / масштаб) і стандартизація по каналу, усі константи рахуються тільки на трейн-частині свого фолда. Датасет розрізано випадково на 10 частин: у кожному прогоні 9 частин навчання (90%) і 1 валідація (10%), тож кожне вікно дістає прогноз рівно один раз. Метрики — середнє по 10 фолдах ± розкид.
точність проти повноти
ROC

⑪ Жива перевірка: моделі на свіжих даних колектора

Дані модель бачить уперше. wclass_live.py читає свіжі секунди з майстер-датасету, ріже їх на вікна по 210 с за тим самим правилом, проганяє через продакшен-моделі й дописує результат кожні 210 секунд. Вхід — log1p(x/0.01), жодного числа ззовні вікна; поріг узятий з файлу моделі. Ціна в модель не входить.
ціна · колесо = масштаб · тягни = зсув

⑪а · Згорткова мережа Conv1D

Conv1D · яскравість = впевненість

⑪б · Градієнтний бустинг

Той самий вхід, той самий час, інший движок. Де вони згодні — вікно однозначне; де ні — воно лежить на межі класу.
бустинг · яскравість = впевненість

⑫ Ті самі моделі на всьому датасеті

Читати з застереженням: 98% цих вікон були в навчанні продакшен-моделей, тож тут переважно памʼять, а не передбачення. Чесна частина — 4 відкладені шматки по 0.5%, вони позначені смугами внизу. Для чесної оцінки на всій історії дивись out-of-fold прогнози в розділі ⑤. Цей графік потрібен для іншого: побачити, як модель поводиться на 171 добі, а не на 10 годинах живого потоку.
ціна BTC · колесо = масштаб · тягни = зсув
Conv1D на всій історії
бустинг на всій історії

⑬ Графіки дослідження

gap-статистика силует і вибір K карта вікон проти нуль-контролю портрети класів
календар
``` ### Маршрути дашборда `web/dashboard.py` — додати поруч з іншими маршрутами ```python @app.route('/wclass') def wclass(): return render_template('wclass.html') # класифікація вікон 210 с СТРОГО за 1680 внутрішніми числами (без зовнішньої норми): континуум чи класи @app.route('/api/wclass/live') def wclass_live(): p = os.path.join(ROOT_DIR, 'media', 'analyst', 'wclass', 'live.json') if not os.path.exists(p): return jsonify(dict(error='живого стану ще немає — запусти python3 wclass_live.py BTC')) with open(p) as f: return app.response_class(f.read(), mimetype='application/json') @app.route('/api/wclass/scan') def wclass_scan(): p = os.path.join(ROOT_DIR, 'media', 'analyst', 'wclass', 'scan.json') if not os.path.exists(p): return jsonify(dict(error='прогону по датасету ще немає — запусти python3 wclass.py scan BTC')) with open(p) as f: return app.response_class(f.read(), mimetype='application/json') @app.route('/api/wclass/') def wclass_coin(coin): if not re.fullmatch(r'[A-Za-z0-9]{1,10}', coin or ''): return jsonify(dict(error='погана монета')), 400 p = os.path.join(ROOT_DIR, 'media', 'analyst', 'wclass', f'{coin.upper()}.json') if not os.path.exists(p): return jsonify(dict(error=f'нема звіту по {coin.upper()} — запусти python3 wclass.py report {coin.upper()} full 16 4')) with open(p) as f: return app.response_class(f.read(), mimetype='application/json') ``` ### Агент живого контуру `~/Library/LaunchAgents/com.at.wclasslive.plist` — крок 210 с = вікно ```xml Label com.at.wclasslive ProgramArguments /Library/Frameworks/Python.framework/Versions/3.11/bin/python3 /Users/at/projects/market_dashboard/wclass_live.py BTC EnvironmentVariables OMP_NUM_THREADS 2 RunAtLoad StandardErrorPath /tmp/wclass_live.log StandardOutPath /tmp/wclass_live.log StartInterval 210 WorkingDirectory /Users/at/projects/market_dashboard ``` Підняти: `launchctl bootstrap gui/$UID ~/Library/LaunchAgents/com.at.wclasslive.plist` ### Запис у сторожа вкладок `test_pages.js` — додати до масиву PAGES ```javascript {file: 'wclass.html', min_draw: 200, routes: [['/api/wclass/', 'media/analyst/wclass/{C}.json']]}, ``` Запуск: `node test_pages.js` — ганяє JS сторінки на реальних даних і ловить порожню сторінку при HTTP 200.