From 91af4c2e937b744114d8c2455b010ed25f15d281 Mon Sep 17 00:00:00 2001 From: momentum dev Date: Fri, 25 Sep 2026 16:23:18 -0500 Subject: [PATCH] feat: nightly Ross-recap learning loop on the owner's local Ollama Why: the pilot in data/ross showed that Ross Cameron's daily recaps, joined to the SIP tape and to what our bot saw, answer "which of his trades could we have taken and why didn't we". It was a set of one-off scripts run by hand; the server-side youtube_claims pipeline has been dead since ~08-24 (YouTube blocks the TrueNAS IP) and its captions fallback hid that. This makes the pilot a nightly, local, cumulative loop without a paid LLM API. youtube_claims/recap/ (python -m youtube_claims.recap.nightly): - channel: yt-dlp flat listing of @DaytradeWarrior, m4a audio, faster-whisper small.en on CPU, pilot-format bundles (meta + our scanner list + transcript). - extract: Ollama structured output (JSON schema) per EXTRACT_SPEC, chunked with overlap for long transcripts and merged. The verbatim rule is enforced in code: every quote must be found (token-level, one-word tolerance) and is snapped to the transcript's words; unquoted items, trades with no surviving quote, prices never spoken, and passed-on names never said are dropped. - llm: endpoint pool over RECAP_OLLAMA_HOSTS (round-robin + failover, per-host model/ctx/think pins) since the owner runs several Ollama boxes. - tape/join/ourside: ported join.py/ourside.py -- tickers are price-verified on the Alpaca SIP tape (plus an order-free check for his own named ticker), then paper (:5433) and live (:5434) event views are attached. Bars cached on disk. - store + schema.sql: idempotent recap_* tables in the transcripts schema, upserted per video_id. report: the pilot's sections + "what changed since last run" against the previous day's metrics JSON. - evaluate: scores any model/prompt against the 36 gold extractions. Model choice (36 gold videos, final code path, P/R): gpt-oss:20b on the workstation GPU, two focused calls, think=low: trades P .84 R .70; tickers P .82 R .77; entry +-2% P .54 R .50; passed-on P .66 R .76; temperature 16/31; kind 30/36; ~49 s/video. tools:latest (4B, server) one call: trades P .81 R .72, tickers P .83 R .85, passed-on P .80 R .75, temperature 20/31, but it invents prices (entry P .42) and occasionally runs away (5-9 min). Prices drive the tape verification, so the more precise local model is the default. Iterations: prompt v2 + spoken-price rescale (699 -> 6.99) + split calls took gpt-oss trades F1 .74 -> .76-.78 and tickers .78 -> .79-.80; think=medium blew the token budget. qwen3-30b-a3b on mason only finished 5 videos (host saturated) -- inconclusive. Also: youtube_claims/transcription.py::_captions_fallback now picks the youtube-transcript-api shape by feature detection instead of retrying the removed get_transcript on any error, and transcribe() reports both the whisper and captions causes -- the IpBlocked error was being masked as AttributeError. No trading code touched. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_recap_extract.py | 206 ++++++++++++ tests/unit/test_recap_join_report.py | 228 ++++++++++++++ tests/unit/test_recap_llm_pool.py | 65 ++++ youtube_claims/recap/__init__.py | 12 + youtube_claims/recap/channel.py | 142 +++++++++ youtube_claims/recap/evaluate.py | 182 +++++++++++ youtube_claims/recap/extract.py | 455 +++++++++++++++++++++++++++ youtube_claims/recap/join.py | 44 +++ youtube_claims/recap/llm.py | 133 ++++++++ youtube_claims/recap/nightly.cmd | 16 + youtube_claims/recap/nightly.py | 197 ++++++++++++ youtube_claims/recap/ourside.py | 112 +++++++ youtube_claims/recap/prompt.py | 126 ++++++++ youtube_claims/recap/quotes.py | 94 ++++++ youtube_claims/recap/report.py | 231 ++++++++++++++ youtube_claims/recap/schema.sql | 94 ++++++ youtube_claims/recap/settings.py | 75 +++++ youtube_claims/recap/store.py | 139 ++++++++ youtube_claims/recap/tape.py | 209 ++++++++++++ youtube_claims/transcription.py | 19 +- 20 files changed, 2775 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_recap_extract.py create mode 100644 tests/unit/test_recap_join_report.py create mode 100644 tests/unit/test_recap_llm_pool.py create mode 100644 youtube_claims/recap/__init__.py create mode 100644 youtube_claims/recap/channel.py create mode 100644 youtube_claims/recap/evaluate.py create mode 100644 youtube_claims/recap/extract.py create mode 100644 youtube_claims/recap/join.py create mode 100644 youtube_claims/recap/llm.py create mode 100644 youtube_claims/recap/nightly.cmd create mode 100644 youtube_claims/recap/nightly.py create mode 100644 youtube_claims/recap/ourside.py create mode 100644 youtube_claims/recap/prompt.py create mode 100644 youtube_claims/recap/quotes.py create mode 100644 youtube_claims/recap/report.py create mode 100644 youtube_claims/recap/schema.sql create mode 100644 youtube_claims/recap/settings.py create mode 100644 youtube_claims/recap/store.py create mode 100644 youtube_claims/recap/tape.py diff --git a/tests/unit/test_recap_extract.py b/tests/unit/test_recap_extract.py new file mode 100644 index 0000000..27be583 --- /dev/null +++ b/tests/unit/test_recap_extract.py @@ -0,0 +1,206 @@ +"""Pure pieces of the Ross recap extractor: verbatim-quote verification, chunking, and the +merge of per-chunk results. The Ollama call itself is exercised by the gold-set evaluation.""" + +from collections import Counter + +from youtube_claims.recap.extract import (chunk_bundle, clean_ticker, fix_spoken_prices, merge_records, + scanner_prices, verify_record) +from youtube_claims.recap.quotes import Transcript, transcript_text_from_bundle + +TX = Transcript( + "So I got in on that dip right there, and I got a nice little trade as it rallied back up. " + "Then we come back up right here looking for a curl through VWAP. It fails. Doesn't happen. " + "I took a small loss right there. GLND. This is not an easy one to day trade because it's very thickly traded. " + "Generally, the right approach that I have is to avoid stocks that are giving us those big topping tails.") + + +# ---------------------------------------------------------------- quotes +def test_exact_quote_is_found_and_snapped_to_transcript_words(): + got = TX.find("looking for a curl through vwap") + assert got == "looking for a curl through VWAP." + + +def test_quote_matching_ignores_punctuation_and_case(): + assert TX.find("It fails, doesn't happen, I took a small loss") == "It fails. Doesn't happen. I took a small loss" + + +def test_one_miscopied_word_in_a_long_quote_is_tolerated(): + q = "This is not an easy one to trade because it's very thickly traded" # dropped "day" + got = TX.find(q) + assert got is not None and got.startswith("This is not an easy one") and "thickly traded" in got + + +def test_invented_quote_is_rejected(): + assert TX.find("I made ten thousand dollars on the open with a huge short squeeze") is None + # a real 40-char prefix with an invented tail is NOT enough (the old prefix check accepted it) + assert TX.find("So I got in on that dip right there, and then I sold the whole thing at twelve dollars") is None + + +def test_too_short_quote_is_not_evidence(): + assert TX.find("it fails") is None + assert TX.find("") is None + + +def test_transcript_text_from_bundle_strips_timestamps(): + b = "video_id: x\nTRANSCRIPT (mm:ss text):\n[00:01] Hello there.\n[00:05] Second line." + assert transcript_text_from_bundle(b) == "Hello there. Second line." + + +# ---------------------------------------------------------------- verification +def _raw(): + return { + "video_kind": "daily_recap", "session_date": "2026-09-24", + "day": {"net_pnl": "$2,992.62", "green": True, "market_temperature": "Cold", + "market_quote": "totally made up sentence about the market being slow today"}, + "trades": [ + {"ticker_heard": "a pus", "ticker_guess": "$apus", "setup": "dip_and_curl", + "setup_quote": "I got in on that dip right there", "entry_price": "5.50", "outcome": "WIN", + "why_entered": [{"reason": "dip", "quote": "So I got in on that dip right there"}, + {"reason": "fake", "quote": "because the level two was stacked with buyers everywhere"}], + "stock_context": {"catalyst": "headline", "catalyst_quote": "a quote that does not exist anywhere here", + "other": ["topping tails", "rocket ship"]}}, + {"ticker_heard": "ZZZ", "ticker_guess": "ZZZ", "setup": "bull_flag", + "setup_quote": "I bought the bull flag at the high of day for a big win", + "why_entered": [{"reason": "x", "quote": "entirely invented reason for entering this trade"}]}, + ], + "passed_on": [ + {"ticker_heard": "glnd", "ticker_guess": "GLND", "reason": "grinder", + "quote": "not an easy one to day trade because it's very thickly traded"}, + {"ticker_heard": "pmax", "ticker_guess": "PMAX", "reason": "halts", + "quote": "you cannot possibly manage your risk on these halting stocks"}, + ], + "rules_stated": [{"rule": "avoid topping tails", + "quote": "avoid stocks that are giving us those big topping tails"}, + {"rule": "fake", "quote": "always buy the first red candle after a halt resume"}], + } + + +def test_verify_drops_unquoted_items_and_snaps_the_rest(): + st = Counter() + rec = verify_record(_raw(), TX, st) + assert [t["ticker_guess"] for t in rec["trades"]] == ["APUS"] # ZZZ had no real quote -> dropped + t = rec["trades"][0] + assert t["entry_price"] is None and t["outcome"] == "win" # 5.50 is never said -> dropped + assert [w["reason"] for w in t["why_entered"]] == ["dip"] + assert t["stock_context"]["catalyst"] is None # its quote was invented + assert t["stock_context"]["other"] == ["topping tails"] + assert [p["ticker_guess"] for p in rec["passed_on"]] == ["GLND"] + assert rec["passed_on"][0]["quote"].startswith("not an easy one to day trade") + assert len(rec["rules_stated"]) == 1 + assert rec["day"]["net_pnl"] == 2992.62 and rec["day"]["market_temperature"] == "cold" + assert rec["day"]["market_quote"] is None + assert st["dropped_trades"] == 1 and st["dropped_passed_on"] == 1 and st["dropped_rules"] == 1 + + +def test_clean_ticker(): + assert clean_ticker("$apus") == "APUS" + assert clean_ticker("K I T T") == "KITT" + assert clean_ticker("null") is None and clean_ticker("Greenland stocks") is None + + +# ---------------------------------------------------------------- chunking +def test_short_bundle_is_one_chunk(): + b = "video_id: x\nTRANSCRIPT (mm:ss text):\n[00:01] hi" + assert chunk_bundle(b, 10_000) == [b] + + +def test_long_bundle_chunks_keep_header_and_overlap(): + head = "video_id: x\nupload_date: 20260924\nTRANSCRIPT (mm:ss text):" + lines = [f"[{i // 60:02d}:{i % 60:02d}] line number {i} " + "x" * 80 for i in range(200)] + b = head + "\n" + "\n".join(lines) + chunks = chunk_bundle(b, 4000, overlap_lines=3) + assert len(chunks) > 3 + assert all(c.startswith(head) for c in chunks) + body = [c.split("TRANSCRIPT (mm:ss text):\n", 1)[1].splitlines() for c in chunks] + assert body[0][-3:] == body[1][:3] # overlap carried over + seen = {ln for part in body for ln in part} + assert seen == set(lines) # nothing lost + + +# ---------------------------------------------------------------- merge +def _trade(tk, setup_q, reasons=(), **kw): + return {"ticker_heard": tk, "ticker_guess": tk, "setup": kw.pop("setup", "other"), "setup_quote": setup_q, + "entry_price": kw.pop("entry_price", None), "exit_price": None, "outcome": kw.pop("outcome", None), + "stock_context": {"catalyst": None, "other": kw.pop("other", [])}, + "why_entered": [{"reason": r, "quote": q} for r, q in reasons], "tape_reads": [], "why_exited": [], + "mistake_or_lesson": []} + + +def test_merge_dedupes_the_overlap_and_unions_fields(): + a = {"video_kind": "daily_recap", "session_date": "2026-09-24", + "day": {"net_pnl": 100.0, "market_temperature": "cold", "market_quote": None}, + "trades": [_trade("APUS", "i bought the dip right there", [("dip", "i bought the dip right there")], + other=["grinder"])], + "passed_on": [{"ticker_guess": "GLND", "ticker_heard": "glnd", "reason": "r", "quote": "q1"}, + {"ticker_guess": "PFSA", "ticker_heard": "pfsa", "reason": "r", "quote": "q2"}], + "rules_stated": [{"rule": "a", "quote": "cut losses quickly and let winners run"}]} + b = {"video_kind": "daily_recap", "session_date": None, + "day": {"net_pnl": None, "market_temperature": "cold", "market_quote": "slow market today"}, + "trades": [_trade("APUS", "i bought the dip right there and it rallied", entry_price=5.5, outcome="win", + other=["sympathy"]), + _trade("PFSA", "curl through vwap", setup="vwap_break")], + "passed_on": [{"ticker_guess": "GLND", "ticker_heard": "glnd", "reason": "r", "quote": "q1 again"}], + "rules_stated": [{"rule": "a", "quote": "cut losses quickly and let winners run"}, + {"rule": "b", "quote": "size down when the market is cold"}]} + m = merge_records([a, b]) + assert [t["ticker_guess"] for t in m["trades"]] == ["APUS", "PFSA"] + apus = m["trades"][0] + assert apus["entry_price"] == 5.5 and apus["outcome"] == "win" # filled from the later chunk + assert apus["stock_context"]["other"] == ["grinder", "sympathy"] + assert [p["ticker_guess"] for p in m["passed_on"]] == ["GLND", "PFSA"] # GLND deduped across chunks + assert len(m["rules_stated"]) == 2 + assert m["day"] == {"net_pnl": 100.0, "market_temperature": "cold", "market_quote": "slow market today"} + assert m["session_date"] == "2026-09-24" + + +def test_merge_keeps_distinct_trades_on_the_same_ticker(): + a = {"video_kind": "daily_recap", "day": {}, "passed_on": [], "rules_stated": [], + "trades": [_trade("APUS", "first entry on the dip at the open")]} + b = {"video_kind": "daily_recap", "day": {}, "passed_on": [], "rules_stated": [], + "trades": [_trade("APUS", "then i got back in for the break through seven")]} + assert len(merge_records([a, b])["trades"]) == 2 + + +def test_merge_temperature_conflict_becomes_mixed(): + parts = [{"video_kind": "daily_recap", "day": {"market_temperature": t}, "trades": [], "passed_on": [], + "rules_stated": []} for t in ("hot", "cold")] + assert merge_records(parts)["day"]["market_temperature"] == "mixed" + + +def test_spoken_prices_are_rescaled_for_small_caps_only(): + b = ("OUR SCANNER LIST (symbol, gap %, price) by session date:\n" + " 2026-06-23: ICCM 80.1% $7.2, BIG 5.0% $150.0\nTRANSCRIPT (mm:ss text):\n") + px = scanner_prices(b) + assert px == {"ICCM": 7.2, "BIG": 150.0} + rec = {"trades": [{"ticker_guess": "ICCM", "entry_price": 699.0, "exit_price": 7.5, "stated_high": None, + "stated_low": None}, + {"ticker_guess": "BIG", "entry_price": 151.0}]} + assert fix_spoken_prices(rec, px) == 1 + assert rec["trades"][0]["entry_price"] == 6.99 and rec["trades"][1]["entry_price"] == 151.0 + + +def test_paragraph_quotes_are_verified_then_cut_to_a_pointer(): + from youtube_claims.recap.extract import MAX_QUOTE_WORDS, _snap + words = [f"w{i}" for i in range(120)] + tx = Transcript(" ".join(words)) + got = _snap(tx, " ".join(words[5:100])) + assert got == " ".join(words[5:5 + MAX_QUOTE_WORDS]) + + +def test_passed_on_names_must_be_said(): + from youtube_claims.recap.extract import name_mentioned + tx = Transcript("Then K I T T popped up and Greenland stocks ran but I passed on PMAX") + assert name_mentioned(tx, "Kit", "KITT") # spelled out + assert name_mentioned(tx, "Greenland stocks", None) # heard phrase + assert name_mentioned(tx, "p max", "PMAX") + assert not name_mentioned(tx, "benf", "BENF") # only on the scanner list + + +def test_prices_must_be_spoken_in_the_transcript(): + from youtube_claims.recap.extract import price_mentioned + tx = Transcript("I got in at 550 and sold at $6.25, the high was 7 and it traded 1,200 shares") + assert price_mentioned(tx, 5.5) and price_mentioned(tx, 6.25) and price_mentioned(tx, 7.0) + assert not price_mentioned(tx, 8.4) + rec = verify_record({"trades": [{"ticker_guess": "ABC", "setup_quote": "I got in at 550 and sold at", + "entry_price": 5.5, "exit_price": 9.1}]}, tx) + assert rec["trades"][0]["entry_price"] == 5.5 and rec["trades"][0]["exit_price"] is None diff --git a/tests/unit/test_recap_join_report.py b/tests/unit/test_recap_join_report.py new file mode 100644 index 0000000..9de35b0 --- /dev/null +++ b/tests/unit/test_recap_join_report.py @@ -0,0 +1,228 @@ +"""Ross recap loop: ticker/price verification against fake tapes, the report diff, the gold +scorer, and the captions-fallback error surfacing.""" + +import datetime as dt +import sys +import types + +import pytest + +from youtube_claims.recap.evaluate import pair_trades, score_video +from youtube_claims.recap.report import compute, diff_metrics, render +from youtube_claims.recap.tape import entry_bucket, entry_time, path_score, resolve, tape_stats + + +def bars(path, start="07:00"): + """Fake 1-min tape from a list of (high, low); open/close at the midpoint, volume 1000.""" + t0 = dt.datetime.fromisoformat(f"2026-09-24T{start}") + return [(t0 + dt.timedelta(minutes=i), (h + l) / 2, h, l, (h + l) / 2, 1000.0) for i, (h, l) in enumerate(path)] + + +RUNNER = bars([(3.0, 2.9), (3.6, 3.0), (4.2, 3.5), (5.1, 4.1), (4.6, 4.0), (4.3, 3.9)]) # 3 -> 5.1 -> 3.9 +FLAT = bars([(10.0, 9.8), (10.1, 9.9), (10.2, 9.9)]) +OTHER_RUNNER = bars([(3.4, 3.3), (3.6, 3.4), (5.3, 3.5), (5.0, 4.6)]) + + +def test_path_score_accepts_the_story_the_tape_can_tell(): + assert path_score(RUNNER, 3.5, None, None) == 0.5 # entry traded, no high named + s = path_score(RUNNER, 3.5, 5.0, None) # entry then a high near 5.0 (5.1: 2%) + assert s is not None and s < 0.025 + assert path_score(RUNNER, 3.5, 5.0, 3.9) is not None # ...then a low near 3.9 + + +def test_path_score_rejects_impossible_stories(): + assert path_score(FLAT, 3.5, None, None) is None # never traded 3.5 + assert path_score(RUNNER, 3.5, 7.0, None) is None # never reached 7 + assert path_score(RUNNER, 3.5, 5.0, 2.0) is None # never came back to 2 + assert path_score([], 3.5, None, None) is None + assert path_score(RUNNER, None, None, None) is None # no price -> nothing to verify + + +def _bars_for(tapes): + return lambda s: tapes.get(s, []) + + +def test_resolve_verifies_his_ticker_when_the_tape_fits(): + item = {"ticker_heard": "a pus", "ticker_guess": "APUS", "entry_price": 3.5, "stated_high": 5.0} + assert resolve(item, _bars_for({"APUS": RUNNER}), ["FLAT"]) == ("APUS", "price_verified") + + +def test_resolve_prefers_his_name_when_levels_fit_out_of_order(): + # he says "squeezed to 5.1 ... it had come from 2.95": the low came FIRST on the tape + item = {"ticker_heard": "apus", "ticker_guess": "APUS", "entry_price": None, "stated_high": 5.1, + "stated_low": 2.95} + other = bars([(5.2, 5.0), (3.1, 2.9)]) # a scanner name with high-then-low in order + assert resolve(item, _bars_for({"APUS": RUNNER, "OTH": other}), ["OTH"]) == ("APUS", "price_verified_loose") + + +def test_resolve_finds_the_unique_scanner_name_when_his_ticker_is_misheard(): + item = {"ticker_heard": "kit", "ticker_guess": "KIT", "entry_price": 3.5, "stated_high": 5.0} + tapes = {"KIT": FLAT, "KITT": RUNNER, "BIG": FLAT} + assert resolve(item, _bars_for(tapes), ["BIG", "KITT"]) == ("KITT", "price_matched_scanner") + + +def test_resolve_reports_ambiguity_instead_of_guessing(): + item = {"ticker_heard": "x", "ticker_guess": "XX", "entry_price": 3.5, "stated_high": 5.1} + got = resolve(item, _bars_for({"AAA": RUNNER, "BBB": RUNNER}), ["AAA", "BBB"]) + assert got[0] is None and got[1].startswith("ambiguous:") and "AAA" in got[1] and "BBB" in got[1] + + +def test_resolve_no_price_match_and_unverified(): + item = {"ticker_heard": "zz", "ticker_guess": "ZZ", "entry_price": 50.0, "stated_high": None} + assert resolve(item, _bars_for({"ZZ": RUNNER}), ["ZZ"]) == ("ZZ", "no_price_match") + item = {"ticker_heard": "zz", "ticker_guess": "ZZ", "entry_price": None, "stated_high": None} + assert resolve(item, _bars_for({}), []) == ("ZZ", "unverified") + assert resolve({"ticker_heard": "Greenland", "ticker_guess": ""}, _bars_for({}), []) == (None, "unverified") + + +def test_entry_time_is_first_touch_before_his_stated_high(): + t = entry_time(RUNNER, 4.1, 5.1) + assert t.strftime("%H:%M") == "07:02" + assert entry_time(RUNNER, 9.9, None) is None + assert entry_bucket(t) == "7:00-9:30" and entry_bucket(dt.datetime(2026, 9, 24, 9, 31)) == "9:30-9:45" + + +def test_tape_stats_splits_premarket_and_rth(): + tape = bars([(2.0, 1.9), (2.8, 2.0)], start="09:28") + bars([(3.0, 2.5), (2.6, 2.2)], start="09:30") + s = tape_stats(tape, prev_close=2.0) + assert s["pm_high"] == 2.8 and s["hod"] == 3.0 and s["hod_time"] == "09:30" + assert s["open"] == pytest.approx(2.75) and s["gap_pct"] == pytest.approx(0.375) + assert s["first_30pct_time"] == "09:29" + + +# ---------------------------------------------------------------- report +def _view(disc=None, orders=None, oop=None): + side = {"discovered": [[disc, "Discovered"]] if disc else [], "orders": orders or [], + "states": [["08:00", f"x out_of_population: price 1.5 < 2"]] if oop else [], + "top_blocks": [["spread too wide", 3]]} + return {"paper": side, "live": None, "tape": {"prev_close": 2.0, "open": 3.0, "close": 3.3, + "open_to_hod_pct": 0.4}} + + +def _rows(n_videos=2): + videos = [{"video_id": f"v{i}", "session_date": f"2026-09-2{i}", "video_kind": "daily_recap", + "extractor": "gold", "market_temperature": "cold"} for i in range(n_videos)] + trades = [{"video_id": "v0", "session_date": "2026-09-20", "symbol": "APUS", "resolve": "price_verified", + "setup": "micro_pullback", "outcome": "win", "pnl": 1200, "entry_price": 3.0, + "entry_time_tape": "07:05", "stock_context": {"catalyst": "headline"}, + "our_view": _view(disc="06:59", orders=[["07:10", "buy"]])}, + {"video_id": "v1", "session_date": "2026-09-21", "symbol": None, "resolve": "ambiguous:A,B", + "setup": "other", "outcome": None, "pnl": None, "entry_price": 5.0, "entry_time_tape": None, + "stock_context": {}, "our_view": None}] + passed = [{"video_id": "v0", "symbol": "GLND", "our_view": _view()}] + return videos, trades, passed + + +def test_compute_aggregates_coverage_and_buckets(): + m = compute(*_rows()) + assert m["videos"] == 2 and m["trades"] == 2 and m["verified"] == 1 + assert m["resolution"] == {"price_verified": 1, "ambiguous": 1} + assert m["entry_buckets"]["7:00-9:30"] == 1 + assert m["coverage"] == {"verified": 1, "discovered": 1, "discovered_before_entry": 1, "we_ordered": 1, + "out_of_population": 0} + assert m["passed_tape"]["with_tape"] == 1 and m["passed_tape"]["median_open_to_close"] == pytest.approx(0.1) + md = render(m, ["x"], today="2026-09-25") + assert "## What changed since last run" in md and "## Our coverage of his verified trades" in md + + +def test_diff_first_run_and_no_change(): + m = compute(*_rows()) + assert diff_metrics(None, m) == ["First report -- no previous run to compare with."] + assert diff_metrics(m, m) == ["No change since the previous run."] + + +def test_diff_reports_new_videos_trades_and_metric_moves(): + prev = compute(*_rows()) + videos, trades, passed = _rows(3) + trades.append({"video_id": "v2", "session_date": "2026-09-22", "symbol": "KITT", "resolve": "price_verified", + "setup": "bull_flag", "outcome": "loss", "pnl": -500, "entry_price": 6.0, + "entry_time_tape": "09:31", "stock_context": {}, "our_view": _view()}) + cur = compute(videos, trades, passed) + lines = diff_metrics(prev, cur) + text = "\n".join(lines) + assert "New videos (1): v2" in text + assert "trades: 2 -> 3 (+1)" in text and "verified: 1 -> 2 (+1)" in text + assert "entry_buckets.9:30-9:45: 0 -> 1" in text + assert "New trades (1): KITT|bull_flag|6.0" in text + + +# ---------------------------------------------------------------- evaluation scorer +def test_score_video_perfect_and_partial(): + gold = {"video_kind": "daily_recap", "day": {"market_temperature": "hot", "net_pnl": 1000}, + "trades": [{"ticker_guess": "APUS", "setup": "micro_pullback", "entry_price": 5.0, "exit_price": 5.5}, + {"ticker_guess": "APUS", "setup": "vwap_break", "entry_price": 4.0, "exit_price": None}, + {"ticker_guess": "PFSA", "setup": "other", "entry_price": None, "exit_price": None}], + "passed_on": [{"ticker_guess": "GLND"}, {"ticker_guess": "PMAX"}], "rules_stated": []} + s = score_video(gold, gold) + assert s["matched_trades"] == 3 and s["setup_ok"] == 3 and s["tick_tp"] == 2 and s["pass_tp"] == 2 + pred = {"video_kind": "daily_recap", "day": {"market_temperature": "cold", "net_pnl": 1010}, + "trades": [{"ticker_guess": "APUS", "setup": "vwap_break", "entry_price": 4.05, "exit_price": 9.0}, + {"ticker_guess": "ZZZ", "setup": "other"}], + "passed_on": [{"ticker_guess": "GLND"}, {"ticker_guess": "BENF"}], "rules_stated": []} + s = score_video(gold, pred) + assert s["matched_trades"] == 1 and s["setup_ok"] == 1 # paired with the vwap_break trade + assert s["entry_price_ok"] == 1 and s["exit_price_ok"] == 0 and s["exit_price_pred"] == 1 + assert (s["tick_tp"], s["tick_pred"], s["tick_gold"]) == (1, 2, 2) + assert (s["pass_tp"], s["pass_pred"], s["pass_gold"]) == (1, 2, 2) + assert s["temp_ok"] == 0 and s["pnl_ok"] == 1 + + +def test_pair_trades_matches_each_trade_once(): + g = [{"ticker_guess": "A", "setup": "x"}] + p = [{"ticker_guess": "A", "setup": "x"}, {"ticker_guess": "A", "setup": "x"}] + assert len(pair_trades(g, p)) == 1 + + +# ---------------------------------------------------------------- captions fallback +def _fake_api(monkeypatch, cls): + mod = types.ModuleType("youtube_transcript_api") + mod.YouTubeTranscriptApi = cls + monkeypatch.setitem(sys.modules, "youtube_transcript_api", mod) + + +class IpBlocked(Exception): + pass + + +def test_captions_fallback_surfaces_the_real_error(monkeypatch): + from youtube_claims import transcription + + class NewApi: # >=1.0 shape, no get_transcript + def fetch(self, vid): + raise IpBlocked("YouTube is blocking requests from your IP") + + _fake_api(monkeypatch, NewApi) + with pytest.raises(IpBlocked, match="blocking requests"): + transcription._captions_fallback("abc") + + +def test_captions_fallback_old_api_still_works(monkeypatch): + from youtube_claims import transcription + + class OldApi: + @staticmethod + def get_transcript(vid): + return [{"start": 1.0, "duration": 2.0, "text": "hi"}] + + _fake_api(monkeypatch, OldApi) + assert transcription._captions_fallback("abc") == [{"start": 1.0, "end": 3.0, "text": "hi"}] + + +def test_transcribe_reports_both_causes(monkeypatch, tmp_path): + from youtube_claims import config, transcription + + class NewApi: + def fetch(self, vid): + raise IpBlocked("blocked") + + _fake_api(monkeypatch, NewApi) + monkeypatch.setattr(config, "TRANSCRIPT_DIR", str(tmp_path)) + + def no_audio(vid, td): + raise RuntimeError("Sign in to confirm you're not a bot") + + monkeypatch.setattr(transcription, "_download_audio", no_audio) + with pytest.raises(RuntimeError) as ei: + transcription.transcribe("abc", watchlist=[]) + msg = str(ei.value) + assert "not a bot" in msg and "IpBlocked: blocked" in msg and "get_transcript" not in msg diff --git a/tests/unit/test_recap_llm_pool.py b/tests/unit/test_recap_llm_pool.py new file mode 100644 index 0000000..09d001a --- /dev/null +++ b/tests/unit/test_recap_llm_pool.py @@ -0,0 +1,65 @@ +"""Round-robin + failover across the owner's Ollama endpoints (no network: fake transport).""" + +import urllib.error + +import pytest + +from youtube_claims.recap.llm import OllamaPool, parse_hosts + + +def test_parse_hosts_with_per_host_model_and_options(): + eps = parse_hosts(" http://a:11434/ , http://b:11434=qwen3:30b#ctx=8192#think=false,, ") + assert [e.host for e in eps] == ["http://a:11434", "http://b:11434"] + assert eps[0].model is None and eps[0].think == "default" + assert eps[1].model == "qwen3:30b" and eps[1].num_ctx == 8192 and eps[1].think is False + + +def _pool(spec, models, fail=(), default="m1"): + calls = [] + + def call(system, user, *, model, host, num_ctx, think, **kw): + calls.append((host, model, num_ctx, think)) + if host in fail: + raise urllib.error.URLError("connection refused") + return "{}", {"seconds": 1} + + pool = OllamaPool(parse_hosts(spec), default, default_ctx=24576, default_think="low", call=call, + list_models=lambda h: models[h]) + return pool, calls + + +def test_round_robin_over_hosts_that_have_the_model(): + pool, calls = _pool("http://a,http://b,http://c", {"http://a": {"m1"}, "http://b": {"other"}, + "http://c": {"m1:latest"}}) + for _ in range(4): + pool.chat("s", "u") + assert [c[0] for c in calls] == ["http://a", "http://c", "http://c", "http://a"] + + +def test_failover_parks_a_dead_host(): + pool, calls = _pool("http://a,http://b", {"http://a": {"m1"}, "http://b": {"m1"}}, fail={"http://a"}) + _, st = pool.chat("s", "u") + assert st["host"] == "http://b" + pool.chat("s", "u") + assert [c[0] for c in calls] == ["http://a", "http://b", "http://b"] # a parked after failing + + +def test_per_host_model_and_options_are_used(): + pool, calls = _pool("http://a=qwen#ctx=8192#think=false", {"http://a": {"qwen"}}) + _, st = pool.chat("s", "u") + assert calls == [("http://a", "qwen", 8192, False)] and st["model"] == "qwen" + + +def test_no_endpoint_with_the_model_is_an_error(): + pool, _ = _pool("http://a", {"http://a": {"x"}}) + with pytest.raises(RuntimeError, match="no reachable Ollama endpoint"): + pool.chat("s", "u") + + +def test_client_errors_are_not_retried_elsewhere(): + def call(*a, **kw): + raise RuntimeError("ollama 400: invalid think value") + + pool = OllamaPool(parse_hosts("http://a,http://b"), "m1", call=call, list_models=lambda h: {"m1"}) + with pytest.raises(RuntimeError, match="400"): + pool.chat("s", "u") diff --git a/youtube_claims/recap/__init__.py b/youtube_claims/recap/__init__.py new file mode 100644 index 0000000..4b74a53 --- /dev/null +++ b/youtube_claims/recap/__init__.py @@ -0,0 +1,12 @@ +"""Ross Cameron daily-recap learning loop. + +A nightly, DESCRIPTIVE pipeline over the Warrior Trading recap videos: list new uploads -> +yt-dlp audio + faster-whisper -> bundle (meta + our scanner list + transcript) -> local Ollama +extraction per EXTRACT_SPEC (verbatim quotes enforced in code) -> price-verify tickers against +the SIP tape + attach what OUR bot saw that day -> Postgres + a cumulative markdown report. + +It records what he SAID and what our system DID; it never changes trading behaviour. + + python -m youtube_claims.recap.nightly # the whole nightly pass + python -m youtube_claims.recap.evaluate # score extraction vs the gold set +""" diff --git a/youtube_claims/recap/channel.py b/youtube_claims/recap/channel.py new file mode 100644 index 0000000..fe760c5 --- /dev/null +++ b/youtube_claims/recap/channel.py @@ -0,0 +1,142 @@ +"""Channel listing, audio download, whisper transcription and bundle building. + +Runs on the owner's Windows workstation: YouTube blocks the TrueNAS server's IP, and the caption +API rate-limits (IpBlocked after ~20 requests), so we pull AUDIO with yt-dlp -- one polite +request per new video -- and transcribe locally with faster-whisper.""" + +from __future__ import annotations + +import datetime as dt +import json +import time +from pathlib import Path + +from youtube_claims.recap import settings + +_whisper = None + + +def list_uploads(limit: int | None = None) -> list[dict]: + """Newest-first [{id, title, duration}] from the channel's /videos tab (flat, one request).""" + import yt_dlp + opts = {"extract_flat": "in_playlist", "quiet": True, "no_warnings": True, + "playlistend": limit or settings.LIST_LIMIT} + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(settings.CHANNEL_URL, download=False) + out = [] + for e in info.get("entries") or []: + if e and e.get("id"): + out.append({"id": e["id"], "title": e.get("title") or "", "duration": e.get("duration")}) + return out + + +def video_meta(vid: str, title: str = "", duration=None) -> dict: + """{id, title, duration, upload_date} -- cached in transcripts/.meta.json (the flat + listing has no upload date, so new ids cost one metadata request each).""" + f = settings.dirs()["transcripts"] / f"{vid}.meta.json" + if f.exists(): + m = json.loads(f.read_text(encoding="utf-8")) + if m.get("upload_date"): + return m + import yt_dlp + with yt_dlp.YoutubeDL({"quiet": True, "no_warnings": True, "skip_download": True}) as ydl: + info = ydl.extract_info(f"https://www.youtube.com/watch?v={vid}", download=False) + m = {"id": vid, "title": info.get("title") or title, "duration": str(info.get("duration") or duration or ""), + "upload_date": info.get("upload_date") or ""} + f.write_text(json.dumps(m), encoding="utf-8") + return m + + +def download_audio(vid: str) -> Path: + """Best m4a audio (no ffmpeg re-encode needed; faster-whisper decodes it directly).""" + ad = settings.dirs()["audio"] + have = next(iter(ad.glob(f"{vid}.*")), None) + if have and not have.name.endswith(".part"): + return have + import yt_dlp + opts = {"format": "ba[ext=m4a]/ba", "outtmpl": str(ad / "%(id)s.%(ext)s"), "quiet": True, + "no_warnings": True, "noprogress": True} + with yt_dlp.YoutubeDL(opts) as ydl: + ydl.download([f"https://www.youtube.com/watch?v={vid}"]) + got = next((p for p in ad.glob(f"{vid}.*") if not p.name.endswith(".part")), None) + if got is None: + raise RuntimeError(f"yt-dlp produced no audio file for {vid}") + return got + + +def transcribe(vid: str, audio: Path) -> list[dict]: + """faster-whisper segments, cached in transcripts/.whisper.json (the pilot's format).""" + out = settings.dirs()["transcripts"] / f"{vid}.whisper.json" + if out.exists(): + return json.loads(out.read_text(encoding="utf-8"))["segments"] + global _whisper + if _whisper is None: + from faster_whisper import WhisperModel + _whisper = WhisperModel(settings.WHISPER_MODEL, device="cpu", compute_type="int8", + cpu_threads=settings.WHISPER_THREADS) + t0 = time.time() + segs, _ = _whisper.transcribe(str(audio), initial_prompt=settings.WHISPER_PROMPT, vad_filter=True, + language="en") + rows = [{"start": round(s.start, 2), "end": round(s.end, 2), "text": s.text} for s in segs] + out.write_text(json.dumps({"video_id": vid, "source": "whisper", "whisper_model": settings.WHISPER_MODEL, + "seconds": round(time.time() - t0, 1), "segments": rows}), encoding="utf-8") + return rows + + +def best_transcript(vid: str) -> tuple[str | None, list[dict] | None]: + """Local whisper (small.en) > repo whisper (medium) > YouTube captions -- as the pilot.""" + td = settings.dirs()["transcripts"] + repo_tx = settings.DATA_DIR.parent / "transcripts" / f"{vid}.json" # youtube_claims whisper-medium + for path, src in ((td / f"{vid}.whisper.json", "whisper-small"), (repo_tx, "whisper-medium")): + if path.exists(): + return src, json.loads(path.read_text(encoding="utf-8"))["segments"] + cap = td / f"{vid}.captions.json" + if cap.exists(): + rows = json.loads(cap.read_text(encoding="utf-8")) + return "captions", [{"start": r["start"], "text": r["text"]} for r in rows] + return None, None + + +def prior_weekdays(upload_date: str, n: int = 1) -> list[str]: + """The upload date (if a weekday) and the n weekdays before it, as YYYY-MM-DD.""" + d = dt.date(int(upload_date[:4]), int(upload_date[4:6]), int(upload_date[6:])) + out = [d.isoformat()] if d.weekday() < 5 else [] + while len(out) < (n + 1 if d.weekday() < 5 else n): + d -= dt.timedelta(days=1) + if d.weekday() < 5: + out.append(d.isoformat()) + return out + + +def scanner_for(upload_date: str, ourside=None) -> dict: + """{session date: [(symbol, gap %, price)]} for the upload day and the weekday before -- + cached in scanner/.json (the pilot's format).""" + f = settings.dirs()["scanner"] / f"{upload_date}.json" + if f.exists(): + return json.loads(f.read_text()) + if ourside is None: + return {} + scan = {d: [list(r) for r in ourside.day_universe(d)] for d in prior_weekdays(upload_date)} + if any(scan.values()): + f.write_text(json.dumps(scan)) + return scan + + +def bundle_source(bundle: str) -> str | None: + for ln in bundle.splitlines()[:6]: + if ln.startswith("transcript_source: "): + return ln.split(": ", 1)[1].strip() + return None + + +def make_bundle(meta: dict, source: str, segments: list[dict], scanner: dict) -> str: + """One plain-text bundle: meta + our scanner list + timestamped transcript (pilot format).""" + lines = [f"video_id: {meta['id']}", f"upload_date: {meta['upload_date']}", f"title: {meta.get('title', '')}", + f"transcript_source: {source}", "", "OUR SCANNER LIST (symbol, gap %, price) by session date:"] + for d, rows in scanner.items(): + lines.append(f" {d}: " + ", ".join(f"{s} {g}% ${p}" if p else f"{s} {g}%" for s, g, p in rows)) + lines += ["", "TRANSCRIPT (mm:ss text):"] + for s in segments: + t = int(s["start"]) + lines.append(f"[{t // 60:02d}:{t % 60:02d}] {s['text'].strip()}") + return "\n".join(lines) diff --git a/youtube_claims/recap/evaluate.py b/youtube_claims/recap/evaluate.py new file mode 100644 index 0000000..b9e396f --- /dev/null +++ b/youtube_claims/recap/evaluate.py @@ -0,0 +1,182 @@ +"""Score the local-model extraction against the GOLD extractions (data/ross/extracted/, done by +a strong model on the same bundles). Per video: trade count, traded tickers, entry/exit price +(+-2%), setup label, passed-on tickers, market temperature, video kind, net P&L. + + python -m youtube_claims.recap.evaluate --run v1 # extract (cached) + score + python -m youtube_claims.recap.evaluate --run v1 --model tools:latest --host http://192.168.1.5:11434 --num-ctx 0 + python -m youtube_claims.recap.evaluate --run v1 --score-only # re-score cached outputs + +Model outputs are cached per run under data/ross/eval// so scoring changes are free.""" + +from __future__ import annotations + +import argparse +import json +import time +from collections import Counter +from pathlib import Path + +from youtube_claims.recap import settings + + +def _close(a, b, tol=0.02) -> bool: + return a is not None and b is not None and abs(a - b) <= tol * max(abs(b), 1e-9) + + +def _tick(x: dict) -> str | None: + return (x.get("ticker_guess") or x.get("ticker_heard") or "").upper().replace(" ", "") or None + + +def pair_trades(gold: list[dict], pred: list[dict]) -> list[tuple[dict, dict]]: + """Match predicted to gold trades on the same ticker, greedily by agreement (setup, entry + price, exit price), ties broken by order. A trade matches at most once.""" + pairs, used = [], set() + cand = [] + for gi, g in enumerate(gold): + for pi, p in enumerate(pred): + if _tick(g) and _tick(g) == _tick(p): + score = (g.get("setup") == p.get("setup")) + _close(p.get("entry_price"), g.get("entry_price")) \ + + _close(p.get("exit_price"), g.get("exit_price")) + cand.append((-score, abs(gi - pi), gi, pi)) + gdone = set() + for _, _, gi, pi in sorted(cand): + if gi in gdone or pi in used: + continue + gdone.add(gi); used.add(pi) + pairs.append((gold[gi], pred[pi])) + return pairs + + +def score_video(gold: dict, pred: dict) -> dict: + gt, pt = gold.get("trades") or [], pred.get("trades") or [] + pairs = pair_trades(gt, pt) + s = Counter({k: 0 for k in ("entry_price_ok", "entry_price_gold", "entry_price_pred", "exit_price_ok", + "exit_price_gold", "exit_price_pred", "setup_ok", "outcome_ok", "temp_gold", + "temp_ok", "pnl_gold", "pnl_ok")}) + s["videos"] = 1 + s["gold_trades"], s["pred_trades"], s["matched_trades"] = len(gt), len(pt), len(pairs) + s["trade_count_exact"] = int(len(gt) == len(pt)) + s["trade_count_abs_err"] = abs(len(gt) - len(pt)) + gts, pts = {_tick(t) for t in gt} - {None}, {_tick(t) for t in pt} - {None} + s["tick_tp"], s["tick_gold"], s["tick_pred"] = len(gts & pts), len(gts), len(pts) + for field in ("entry_price", "exit_price"): + for g, p in pairs: + gv, pv = g.get(field), p.get(field) + if gv is not None: + s[f"{field}_gold"] += 1 + if pv is not None: + s[f"{field}_pred"] += 1 + if _close(pv, gv): + s[f"{field}_ok"] += 1 + for g, p in pairs: + s["setup_ok"] += int(g.get("setup") == p.get("setup")) + s["outcome_ok"] += int(g.get("outcome") == p.get("outcome")) + gp = {_tick(x) for x in gold.get("passed_on") or []} - {None} + pp = {_tick(x) for x in pred.get("passed_on") or []} - {None} + s["pass_tp"], s["pass_gold"], s["pass_pred"] = len(gp & pp), len(gp), len(pp) + gtemp = (gold.get("day") or {}).get("market_temperature") + ptemp = (pred.get("day") or {}).get("market_temperature") + if gtemp: + s["temp_gold"] += 1 + s["temp_ok"] += int(gtemp == ptemp) + s["kind_ok"] = int(gold.get("video_kind") == pred.get("video_kind")) + gp_, pp_ = (gold.get("day") or {}).get("net_pnl"), (pred.get("day") or {}).get("net_pnl") + if gp_ is not None: + s["pnl_gold"] += 1 + s["pnl_ok"] += int(_close(pp_, gp_)) + s["rules_gold"], s["rules_pred"] = len(gold.get("rules_stated") or []), len(pred.get("rules_stated") or []) + return dict(s) + + +def _pr(tp, npred, ngold) -> str: + p = tp / npred if npred else 0.0 + r = tp / ngold if ngold else 0.0 + f = 2 * p * r / (p + r) if p + r else 0.0 + return f"P {p:.2f} R {r:.2f} F1 {f:.2f} ({tp}/{npred} pred, {tp}/{ngold} gold)" + + +def summarize(per_video: dict[str, dict]) -> tuple[str, dict]: + tot = Counter() + for s in per_video.values(): + tot.update(s) + n = tot["videos"] or 1 + lines = [ + f"videos scored: {tot['videos']}", + f"trades (matched on ticker): {_pr(tot['matched_trades'], tot['pred_trades'], tot['gold_trades'])}", + f"trade count exact: {tot['trade_count_exact']}/{tot['videos']} mean |err| {tot['trade_count_abs_err'] / n:.2f}", + f"traded tickers: {_pr(tot['tick_tp'], tot['tick_pred'], tot['tick_gold'])}", + f"entry price +-2% (matched trades): {_pr(tot['entry_price_ok'], tot['entry_price_pred'], tot['entry_price_gold'])}", + f"exit price +-2% (matched trades): {_pr(tot['exit_price_ok'], tot['exit_price_pred'], tot['exit_price_gold'])}", + f"setup label agrees: {tot['setup_ok']}/{tot['matched_trades']} outcome agrees: {tot['outcome_ok']}/{tot['matched_trades']}", + f"passed-on tickers: {_pr(tot['pass_tp'], tot['pass_pred'], tot['pass_gold'])}", + f"market temperature: {tot['temp_ok']}/{tot['temp_gold']} video kind: {tot['kind_ok']}/{tot['videos']}", + f"net P&L +-2%: {tot['pnl_ok']}/{tot['pnl_gold']} rules: {tot['rules_pred']} pred vs {tot['rules_gold']} gold", + ] + return "\n".join(lines), dict(tot) + + +def main() -> None: + ap = argparse.ArgumentParser(description="Score recap extraction vs the gold set") + ap.add_argument("--run", required=True, help="name of this configuration (cache dir)") + ap.add_argument("--model") + ap.add_argument("--host") + ap.add_argument("--num-ctx", type=int, help="0 = the model's default context") + ap.add_argument("--think", help="false|true|low|medium|high (model-dependent)") + ap.add_argument("--chunk-chars", type=int) + ap.add_argument("--system-file", help="alternative system prompt (prompt iteration)") + ap.add_argument("--split", choices=["0", "1"], help="two focused calls per chunk (default: settings)") + ap.add_argument("--limit", type=int) + ap.add_argument("--only", help="comma-separated substrings of bundle names") + ap.add_argument("--score-only", action="store_true") + a = ap.parse_args() + from youtube_claims.recap.extract import extract_bundle + d = settings.dirs() + gold_dir, bundles = d["extracted"], d["bundles"] + out_dir = settings.DATA_DIR / "eval" / a.run + out_dir.mkdir(parents=True, exist_ok=True) + think = None if a.think is None else {"false": False, "true": True}.get(a.think.lower(), a.think) + golds = sorted(gold_dir.glob("*.json")) + if a.only: + keys = a.only.split(",") + golds = [g for g in golds if any(k in g.stem for k in keys)] + if a.limit: + golds = golds[:a.limit] + per, secs = {}, [] + for g in golds: + cache = out_dir / g.name + if not cache.exists() and not a.score_only: + b = (bundles / f"{g.stem}.txt").read_text(encoding="utf-8") + t0 = time.time() + try: + rec, run = extract_bundle(b, model=a.model, host=a.host, num_ctx=a.num_ctx, think=think, + chunk_chars=a.chunk_chars, + split=None if a.split is None else a.split == "1", + system=Path(a.system_file).read_text(encoding="utf-8") + if a.system_file else None) + except Exception as e: # noqa: BLE001 -- score the rest; a failure scores as empty + print(g.stem, "FAILED:", e, flush=True) + rec, run = {"trades": [], "passed_on": [], "rules_stated": [], "day": {}}, {"error": str(e)} + run["wall_seconds"] = round(time.time() - t0, 1) + cache.write_text(json.dumps({**rec, "_run": run}, indent=1), encoding="utf-8") + if not cache.exists(): + continue + pred = json.loads(cache.read_text(encoding="utf-8")) + gold = json.loads(g.read_text(encoding="utf-8")) + per[g.stem] = score_video(gold, pred) + r = pred.get("_run") or {} + if r.get("wall_seconds"): + secs.append(r["wall_seconds"]) + s = per[g.stem] + print(f"{g.stem}: trades {s['pred_trades']}/{s['gold_trades']} matched {s['matched_trades']} " + f"tickers {s['tick_tp']}/{s['tick_gold']} passed {s['pass_tp']}/{s['pass_gold']} " + f"temp {'ok' if s.get('temp_ok') else '-'} {r.get('wall_seconds', '?')}s drops {r.get('stats', {})}", + flush=True) + text, tot = summarize(per) + if secs: + text += f"\nextraction seconds/video: mean {sum(secs) / len(secs):.0f}, max {max(secs):.0f}" + print("\n" + text) + (out_dir / "_summary.json").write_text(json.dumps({"totals": tot, "per_video": per, "text": text}, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/youtube_claims/recap/extract.py b/youtube_claims/recap/extract.py new file mode 100644 index 0000000..4498bd0 --- /dev/null +++ b/youtube_claims/recap/extract.py @@ -0,0 +1,455 @@ +"""Bundle -> structured recap record with the owner's local Ollama. + +Pipeline per video: split the bundle into overlapping transcript chunks (only if it is longer +than the model comfortably handles) -> one structured-output call per chunk -> verify every +quote against the FULL transcript (snap to verbatim, drop what can't be found) -> merge the +chunk records (dedupe the overlap) -> normalise tickers. The verification + merge are pure and +unit-tested; only `ollama_chat` touches the network.""" + +from __future__ import annotations + +import json +import re +import time +import urllib.error +import urllib.request +from collections import Counter + +from youtube_claims.recap import prompt as P +from youtube_claims.recap import settings +from youtube_claims.recap.quotes import Transcript, _tok, transcript_text_from_bundle + +_HEADER_MARK = "TRANSCRIPT (mm:ss text):" +_TICKER = re.compile(r"[A-Z]{1,5}") +_NULLISH = {"", "NULL", "NONE", "N/A", "NA", "UNKNOWN", "TBD", "?", "-"} + + +# ---------------------------------------------------------------- Ollama +def ollama_chat(system: str, user: str, *, model: str, host: str, schema: dict | None = None, + num_ctx: int | None = None, think=None, timeout: int = 1800, + temperature: float = 0.0, num_predict: int = 12000) -> tuple[str, dict]: + """One non-streaming /api/chat call in structured-output mode. Returns (content, stats).""" + payload = {"model": model, "stream": False, "format": schema or "json", + "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}], + "options": {"temperature": temperature, "num_predict": num_predict}} + if num_ctx: + payload["options"]["num_ctx"] = num_ctx + if think is not None: + payload["think"] = think + req = urllib.request.Request(host.rstrip("/") + "/api/chat", data=json.dumps(payload).encode(), + headers={"Content-Type": "application/json"}) + t0 = time.time() + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + body = json.load(r) + except urllib.error.HTTPError as e: # surface Ollama's own message (bad model, bad think...) + raise RuntimeError(f"ollama {e.code}: {e.read().decode(errors='replace')[:300]}") from e + stats = {"seconds": round(time.time() - t0, 1), "prompt_tokens": body.get("prompt_eval_count"), + "output_tokens": body.get("eval_count"), "done_reason": body.get("done_reason")} + return (body.get("message") or {}).get("content") or "", stats + + +def parse_json(raw: str) -> dict: + """Structured mode returns bare JSON; tolerate a stray code fence or preamble anyway.""" + raw = (raw or "").strip() + try: + return json.loads(raw) + except json.JSONDecodeError: + m = re.search(r"\{.*\}", raw, re.S) + if not m: + raise + return json.loads(m.group(0)) + + +# ---------------------------------------------------------------- chunking +def split_bundle(bundle: str) -> tuple[str, list[str]]: + """(header incl. the transcript marker, transcript lines).""" + if _HEADER_MARK not in bundle: + return "", [ln for ln in bundle.splitlines() if ln.strip()] + head, body = bundle.split(_HEADER_MARK, 1) + return head + _HEADER_MARK, [ln for ln in body.splitlines() if ln.strip()] + + +def chunk_bundle(bundle: str, max_chars: int, overlap_lines: int = 4) -> list[str]: + """The bundle itself if it fits, else header + consecutive transcript slices of about + max_chars each, consecutive slices sharing `overlap_lines` lines so nothing said across + a boundary is lost (the merge dedupes what both halves extracted).""" + if len(bundle) <= max_chars: + return [bundle] + head, lines = split_bundle(bundle) + budget = max(2000, max_chars - len(head)) + out, i = [], 0 + while i < len(lines): + n, j = 0, i + while j < len(lines) and (n + len(lines[j]) + 1 <= budget or j == i): + n += len(lines[j]) + 1 + j += 1 + out.append(head + "\n" + "\n".join(lines[i:j])) + if j >= len(lines): + break + i = max(i + 1, j - overlap_lines) + return out + + +# ---------------------------------------------------------------- normalisation +def clean_ticker(raw) -> str | None: + t = re.sub(r"[\s.$]", "", str(raw or "")).upper() + if t in _NULLISH or not _TICKER.fullmatch(t): + return None + return t + + +def _num(x): + if x is None or isinstance(x, bool): + return None + if isinstance(x, (int, float)): + return float(x) + m = re.search(r"-?\d[\d,]*\.?\d*", str(x)) + if not m: + return None + try: + return float(m.group(0).replace(",", "")) + except ValueError: + return None + + +def _price_forms(p: float) -> set[str]: + """How a spoken price shows up in a whisper transcript: 5.5 -> 5.50, 5.5, 550, $5.50; + 6.0 -> 6, 6.00, 600; 12.25 -> 12.25, 1225.""" + cents = int(round(p * 100)) + forms = {f"{p:.2f}", f"{p:.2f}".rstrip("0").rstrip("."), str(cents)} + if cents % 100 == 0: + forms.add(str(cents // 100)) + if p < 1: + forms.add(str(cents)) # "73 cents" + return forms + + +def price_mentioned(tx: Transcript, p) -> bool: + """Anti-hallucination for numbers: the price (in some spoken form) occurs in the transcript.""" + if p is None: + return True + toks = tx.token_set + return any(f in toks for f in _price_forms(float(p))) + + +def name_mentioned(tx: Transcript, heard, guess) -> bool: + """The stock was actually named: the heard name (as a phrase) or the guessed symbol occurs + in the transcript, or the symbol spelled out letter by letter ("K I T T").""" + ht = _tok(str(heard or "")) + if ht and (" " + " ".join(ht) + " ") in tx.joined: + return True + g = (str(guess or "")).lower() + return bool(g) and (g in tx.token_set or (" " + " ".join(g) + " ") in tx.joined) + + +def _enum(x, allowed, default=None): + x = (str(x).strip().lower() if x is not None else None) + return x if x in allowed else default + + +# ---------------------------------------------------------------- verification +MAX_QUOTE_WORDS = 50 + + +def _snap(tx: Transcript, q) -> str | None: + """The transcript's own words for quote q (None if not found). A model sometimes pastes a + whole paragraph as the "quote"; it is verified in full, then cut to its first + MAX_QUOTE_WORDS words -- still verbatim, and short enough to stay a pointer, not a copy.""" + s = tx.find(q) if isinstance(q, str) and q.strip() else None + if s is not None: + words = s.split() + if len(words) > MAX_QUOTE_WORDS: + s = " ".join(words[:MAX_QUOTE_WORDS]) + return s + + +def _quoted_list(tx: Transcript, items, key: str, stats: Counter, label: str) -> list[dict]: + out = [] + for it in items or []: + if not isinstance(it, dict): + continue + s = _snap(tx, it.get("quote")) + if s is None: + stats[f"dropped_{label}"] += 1 + continue + out.append({key: str(it.get(key) or "").strip(), "quote": s}) + return out + + +def _descriptor_ok(tx: Transcript, s: str) -> bool: + toks = _tok(s) + if not toks: + return False + if len(toks) >= 3: + return tx.find(s) is not None + return (" " + " ".join(toks) + " ") in tx.joined + + +def verify_record(rec: dict, tx: Transcript, stats: Counter | None = None) -> dict: + """Enforce the verbatim-quote rule in code. Quoted items whose quote is not in the + transcript are dropped; accepted quotes are replaced by the transcript's exact words; + a trade with no surviving quote at all (no evidence he described it) is dropped.""" + stats = stats if stats is not None else Counter() + day_in = rec.get("day") or {} + mq = _snap(tx, day_in.get("market_quote")) + if day_in.get("market_quote") and mq is None: + stats["dropped_market_quote"] += 1 + day = { + "net_pnl": _num(day_in.get("net_pnl")), + "green": day_in.get("green") if isinstance(day_in.get("green"), bool) else None, + "market_temperature": _enum(day_in.get("market_temperature"), {"hot", "cold", "mixed"}), + "market_quote": mq, + "max_loss_hit": day_in.get("max_loss_hit") if isinstance(day_in.get("max_loss_hit"), bool) else None, + "daily_goal_comment": day_in.get("daily_goal_comment") or None, + "sizing_comment": day_in.get("sizing_comment") or None, + } + trades = [] + for t in rec.get("trades") or []: + if not isinstance(t, dict): + continue + sc_in = t.get("stock_context") or {} + cq = _snap(tx, sc_in.get("catalyst_quote")) + sc = {"catalyst": (sc_in.get("catalyst") or None) if cq else None, + "catalyst_quote": cq, + "float": sc_in.get("float") or None, + "pct_up_at_entry": sc_in.get("pct_up_at_entry") or None, + "leading_gainer_rank": _num(sc_in.get("leading_gainer_rank")), + "other": [s for s in (sc_in.get("other") or []) if isinstance(s, str) and _descriptor_ok(tx, s)]} + out = { + "ticker_heard": str(t.get("ticker_heard") or "").strip() or None, + "ticker_guess": clean_ticker(t.get("ticker_guess")) or clean_ticker(t.get("ticker_heard")), + "entry_time_hint": t.get("entry_time_hint") or None, + "setup": _enum(t.get("setup"), set(P.SETUPS), "other"), + "setup_quote": _snap(tx, t.get("setup_quote")), + "entry_price": _num(t.get("entry_price")), "exit_price": _num(t.get("exit_price")), + "stated_high": _num(t.get("stated_high")), "stated_low": _num(t.get("stated_low")), + "shares": _num(t.get("shares")), "pnl": _num(t.get("pnl")), + "outcome": _enum(t.get("outcome"), {"win", "loss", "scratch"}), + "add_on": bool(t.get("add_on")), + "stock_context": sc, + "why_entered": _quoted_list(tx, t.get("why_entered"), "reason", stats, "why_entered"), + "tape_reads": _quoted_list(tx, t.get("tape_reads"), "read", stats, "tape_reads"), + "why_exited": _quoted_list(tx, t.get("why_exited"), "reason", stats, "why_exited"), + "mistake_or_lesson": _quoted_list(tx, t.get("mistake_or_lesson"), "lesson", stats, "lessons"), + } + for k in ("entry_price", "exit_price", "stated_high", "stated_low"): + if out[k] is not None and not price_mentioned(tx, out[k]): + stats["dropped_unspoken_price"] += 1 + out[k] = None + if t.get("setup_quote") and out["setup_quote"] is None: + stats["dropped_setup_quote"] += 1 + if not (out["setup_quote"] or out["why_entered"] or out["why_exited"] or out["tape_reads"] + or out["mistake_or_lesson"]): + stats["dropped_trades"] += 1 + continue + trades.append(out) + passed = [] + for p in rec.get("passed_on") or []: + if not isinstance(p, dict): + continue + s = _snap(tx, p.get("quote")) + if s is None: + stats["dropped_passed_on"] += 1 + continue + if not name_mentioned(tx, p.get("ticker_heard"), p.get("ticker_guess")): + stats["dropped_passed_unnamed"] += 1 # e.g. a scanner-list name he never said + continue + passed.append({"ticker_heard": str(p.get("ticker_heard") or "").strip() or None, + "ticker_guess": clean_ticker(p.get("ticker_guess")) or clean_ticker(p.get("ticker_heard")), + "reason": str(p.get("reason") or "").strip(), "quote": s}) + rules = _quoted_list(tx, rec.get("rules_stated"), "rule", stats, "rules") + return {"video_kind": _enum(rec.get("video_kind"), set(P.KINDS), "other"), + "session_date": rec.get("session_date") if _is_date(rec.get("session_date")) else None, + "day": day, "trades": trades, "passed_on": passed, "rules_stated": rules} + + +def _is_date(s) -> bool: + return isinstance(s, str) and bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", s)) + + +# ---------------------------------------------------------------- merge +def _qkey(q) -> str: + return " ".join(_tok(q or "")) + + +def _overlaps(a: str, b: str) -> bool: + return bool(a) and bool(b) and (a in b or b in a) + + +def _same_trade(a: dict, b: dict) -> bool: + if (a.get("ticker_guess") or a.get("ticker_heard")) != (b.get("ticker_guess") or b.get("ticker_heard")): + return False + qa = {_qkey(a.get("setup_quote"))} | {_qkey(x["quote"]) for k in ("why_entered", "why_exited") for x in a[k]} + qb = {_qkey(b.get("setup_quote"))} | {_qkey(x["quote"]) for k in ("why_entered", "why_exited") for x in b[k]} + qa.discard(""); qb.discard("") + return any(_overlaps(x, y) for x in qa for y in qb) + + +def _union_quoted(a: list[dict], b: list[dict]) -> list[dict]: + out = list(a) + for x in b: + if not any(_overlaps(_qkey(x["quote"]), _qkey(y["quote"])) for y in out): + out.append(x) + return out + + +def merge_records(parts: list[dict]) -> dict: + """Merge verified per-chunk records into one. Chunks overlap, so the same trade / pass / + rule can come back twice: trades are the same when ticker matches and a quote overlaps + (their fields are then unioned, first non-null wins); passes dedupe by ticker; rules by + quote. (A traded ticker may also be passed on -- he often skips a second entry.)""" + parts = [p for p in parts if p] + if not parts: + return {"video_kind": "other", "session_date": None, "day": {}, "trades": [], "passed_on": [], + "rules_stated": []} + kinds = Counter(p.get("video_kind") for p in parts) + kind = max(kinds, key=lambda k: (kinds[k], -[p.get("video_kind") for p in parts].index(k))) + day: dict = {} + for p in parts: + for k, v in (p.get("day") or {}).items(): + if day.get(k) is None and v is not None: + day[k] = v + temps = Counter(p["day"].get("market_temperature") for p in parts if (p.get("day") or {}).get("market_temperature")) + if len(temps) > 1: + top = temps.most_common() + day["market_temperature"] = top[0][0] if top[0][1] > top[1][1] else "mixed" + trades: list[dict] = [] + for p in parts: + for t in p.get("trades") or []: + dup = next((x for x in trades if _same_trade(x, t)), None) + if dup is None: + trades.append(json.loads(json.dumps(t))) + continue + for k, v in t.items(): + if k in ("why_entered", "tape_reads", "why_exited", "mistake_or_lesson"): + dup[k] = _union_quoted(dup[k], v) + elif k == "stock_context": + for ck, cv in (v or {}).items(): + if ck == "other": + dup[k]["other"] = list(dict.fromkeys((dup[k].get("other") or []) + (cv or []))) + elif dup[k].get(ck) is None: + dup[k][ck] = cv + elif dup.get(k) is None and v is not None: + dup[k] = v + passed, seen = [], set() + for p in parts: + for x in p.get("passed_on") or []: + key = x.get("ticker_guess") or (x.get("ticker_heard") or "").upper() + if key in seen: + continue + seen.add(key) + passed.append(x) + rules: list[dict] = [] + for p in parts: + rules = _union_quoted(rules, p.get("rules_stated") or []) + return {"video_kind": kind, "session_date": next((p["session_date"] for p in parts if p.get("session_date")), None), + "day": day, "trades": trades, "passed_on": passed, "rules_stated": rules} + + +# ---------------------------------------------------------------- spoken prices +def scanner_prices(bundle: str) -> dict[str, float]: + """{symbol: price} from the bundle's OUR SCANNER LIST lines ('SYM 12.3% $4.5').""" + out: dict[str, float] = {} + head = bundle.split(_HEADER_MARK, 1)[0] + for sym, px in re.findall(r"\b([A-Z]{1,5}) -?[\d.]+% \$([\d.]+)", head): + out[sym] = max(out.get(sym, 0.0), float(px)) + return out + + +def fix_spoken_prices(rec: dict, prices: dict[str, float]) -> int: + """ASR/LLM render "six ninety-nine" as 699 -- Ross trades small caps, so a trade price + >= 100 is divided by 100 unless the scanner saw that ticker above $50. Returns #fixes.""" + n = 0 + for t in rec.get("trades") or []: + if prices.get(t.get("ticker_guess") or "", 0) >= 50: + continue + for k in ("entry_price", "exit_price", "stated_high", "stated_low"): + v = t.get(k) + if v is not None and v >= 100: + t[k] = round(v / 100, 4) + n += 1 + return n + + +# ---------------------------------------------------------------- driver +def bundle_meta(bundle: str) -> dict: + meta = {} + for ln in bundle.splitlines()[:6]: + k, _, v = ln.partition(": ") + if k in ("video_id", "upload_date", "title", "transcript_source"): + meta[k] = v.strip() + return meta + + +def extract_bundle(bundle: str, *, model: str | None = None, host: str | None = None, + num_ctx: int | None = None, think=None, chunk_chars: int | None = None, + system: str | None = None, split: bool | None = None, log=print) -> tuple[dict, dict]: + """Extract one video. Returns (record per EXTRACT_SPEC, run metadata incl. drop stats). + split=True asks for trades/day and for passed-on/rules in two focused calls per chunk + (better recall from a small local model at ~2x the time).""" + split = settings.SPLIT_CALLS if split is None else split + think = settings.OLLAMA_THINK if think is None else think + from youtube_claims.recap.llm import Endpoint, OllamaPool, from_settings + if host: # an explicit endpoint (evaluation runs) + pool = OllamaPool([Endpoint(host.rstrip("/"))], model or settings.OLLAMA_MODEL, + settings.OLLAMA_NUM_CTX if num_ctx is None else (num_ctx or None), think, + call=ollama_chat) + else: + pool = from_settings() + if model: + pool.default_model = model + if num_ctx is not None: + pool.default_ctx = num_ctx or None + pool.default_think = think + model = pool.default_model + chunk_chars = chunk_chars or settings.CHUNK_CHARS + meta = bundle_meta(bundle) + tx = Transcript(transcript_text_from_bundle(bundle)) + chunks = chunk_bundle(bundle, chunk_chars) + stats: Counter = Counter() + parts, calls = [], [] + tasks = P.SPLIT_TASKS if split else [(None, P.SCHEMA)] + for i, ch in enumerate(chunks, 1): + note = P.PART_NOTE.format(i=i, n=len(chunks)) if len(chunks) > 1 else "" + data: dict | None = {} + for task, schema in tasks: + user = P.USER_TMPL.format(part_note=note + (task or ""), bundle=ch) + got = None + for attempt in range(2): + try: + raw, st = pool.chat(system or P.SYSTEM, user, schema=schema, timeout=settings.OLLAMA_TIMEOUT, + temperature=0.0 if attempt == 0 else 0.3, + num_predict=settings.OLLAMA_NUM_PREDICT) + calls.append(st) + got = parse_json(raw) + break + except (json.JSONDecodeError, ValueError) as e: + stats["json_errors"] += 1 + log(f" chunk {i}/{len(chunks)} bad JSON ({e}); retrying" if attempt == 0 else + f" chunk {i}/{len(chunks)} bad JSON again; skipping") + if got is None: + data = None + break + data.update({k: v for k, v in got.items() if k in schema["properties"]}) # a call may echo other keys empty + if data is None: + stats["chunks_failed"] += 1 + continue + for k in ("trades", "passed_on", "rules_stated"): + stats[f"raw_{k}"] += len(data.get(k) or []) + parts.append(verify_record(data, tx, stats)) + if not parts: + raise RuntimeError(f"extraction produced nothing for {meta.get('video_id')} ({dict(stats)})") + rec = merge_records(parts) + fixed = fix_spoken_prices(rec, scanner_prices(bundle)) + if fixed: + stats["prices_rescaled"] += fixed + up = meta.get("upload_date") or "" + out = {"video_id": meta.get("video_id"), "upload_date": up, + "session_date": rec.get("session_date"), "video_kind": rec["video_kind"], + "day": rec["day"], "trades": rec["trades"], "passed_on": rec["passed_on"], + "rules_stated": rec["rules_stated"]} + run = {"model": model, "hosts": sorted({c.get("host") for c in calls if c.get("host")}), + "chunks": len(chunks), "split": split, "calls": calls, + "seconds": round(sum(c["seconds"] for c in calls), 1), "stats": dict(stats)} + return out, run diff --git a/youtube_claims/recap/join.py b/youtube_claims/recap/join.py new file mode 100644 index 0000000..73dc8ce --- /dev/null +++ b/youtube_claims/recap/join.py @@ -0,0 +1,44 @@ +"""Join one extraction record to OUR data: price-verify each ticker against the SIP tape, locate +his entry on the tape, and attach what our bot saw/did that day (both stacks).""" + +from __future__ import annotations + +from youtube_claims.recap.tape import entry_time, resolve + + +def session_for(rec: dict) -> str: + sd = rec.get("session_date") + if sd: + return str(sd) + u = str(rec.get("upload_date") or "") + return f"{u[:4]}-{u[4:6]}-{u[6:]}" + + +def join_record(rec: dict, ourside, scanner: dict) -> tuple[list[dict], list[dict]]: + """-> (trades, passed): the record's items plus symbol, resolve, our_view (and, for trades, + entry_time_tape HH:MM ET).""" + d = session_for(rec) + scanner_syms = [row[0] for rows in (scanner or {}).values() for row in rows] + bars_for = lambda s: ourside.bars.bars(s, d) # noqa: E731 + views: dict = {} + + def view(sym): + if sym and sym not in views: + try: + views[sym] = ourside.view(d, sym) + except Exception as e: # noqa: BLE001 -- one bad lookup shouldn't sink the video + views[sym] = {"error": str(e)[:200]} + return views.get(sym) + + trades = [] + for t in rec.get("trades") or []: + sym, how = resolve(t, bars_for, scanner_syms) + et = entry_time(bars_for(sym), t.get("entry_price"), t.get("stated_high")) \ + if sym and how in ("price_verified", "price_verified_loose", "price_matched_scanner") else None + trades.append({**t, "symbol": sym, "resolve": how, "our_view": view(sym), + "entry_time_tape": et.strftime("%H:%M") if et else None}) + passed = [] + for p in rec.get("passed_on") or []: + sym, how = resolve(p, bars_for, scanner_syms) + passed.append({**p, "symbol": sym, "resolve": how, "our_view": view(sym)}) + return trades, passed diff --git a/youtube_claims/recap/llm.py b/youtube_claims/recap/llm.py new file mode 100644 index 0000000..790fa5c --- /dev/null +++ b/youtube_claims/recap/llm.py @@ -0,0 +1,133 @@ +"""A small pool over the owner's several LAN Ollama endpoints: round-robin across the endpoints +that actually have the requested model, fail over on connection errors / timeouts / 5xx, and +park a dead endpoint for a few minutes. Endpoints come from RECAP_OLLAMA_HOSTS: + + RECAP_OLLAMA_HOSTS="http://127.0.0.1:11434, http://192.168.1.178:11434=qwen3:30b-a3b-instruct-2507-q4_K_M" + +An entry may pin its own model with `=model` (and `#ctx=N` / `#think=V` options), otherwise it +serves RECAP_OLLAMA_MODEL with the global options. The transport (`call`) is injected so the +rotation/failover logic is unit-tested without a network.""" + +from __future__ import annotations + +import json +import socket +import threading +import time +import urllib.error +import urllib.request +from dataclasses import dataclass, field +from typing import Callable + +_THINK = {"false": False, "true": True, "none": None} + + +@dataclass +class Endpoint: + host: str + model: str | None = None # None -> the pool's default model + num_ctx: int | None = None # None -> the pool's default + think: object = "default" # "default" -> the pool's default + down_until: float = 0.0 + models: set | None = None + models_at: float = 0.0 + + +def parse_hosts(spec: str) -> list[Endpoint]: + """'h1, h2=model#ctx=8192#think=false' -> endpoints (blank entries ignored).""" + out = [] + for raw in (spec or "").split(","): + raw = raw.strip() + if not raw: + continue + head, *opts = raw.split("#") + host, _, model = head.partition("=") + ep = Endpoint(host=host.strip().rstrip("/"), model=model.strip() or None) + for o in opts: + k, _, v = o.partition("=") + k, v = k.strip().lower(), v.strip() + if k == "ctx" and v.isdigit(): + ep.num_ctx = int(v) or None + elif k == "think": + ep.think = _THINK.get(v.lower(), v) + out.append(ep) + return out + + +def _tags(host: str) -> set[str]: + with urllib.request.urlopen(host + "/api/tags", timeout=5) as r: + return {m["name"] for m in json.load(r).get("models", [])} + + +@dataclass +class OllamaPool: + endpoints: list[Endpoint] + default_model: str + default_ctx: int | None = None + default_think: object = None + call: Callable | None = None # (system, user, model=, host=, ...) -> (str, stats) + list_models: Callable[[str], set] = _tags + down_seconds: float = 300.0 + tags_ttl: float = 300.0 + _rr: int = 0 + _lock: threading.Lock = field(default_factory=threading.Lock) + + def _serves(self, ep: Endpoint) -> str | None: + """The model this endpoint would run, if it has it (tags cached); None otherwise.""" + want = ep.model or self.default_model + now = time.monotonic() + if ep.models is None or now - ep.models_at > self.tags_ttl: + try: + ep.models, ep.models_at = self.list_models(ep.host), now + except Exception: # noqa: BLE001 -- unreachable: park it + ep.down_until = now + self.down_seconds + ep.models, ep.models_at = None, 0.0 + return None + names = ep.models or set() + return want if (want in names or f"{want}:latest" in names) else None + + def candidates(self) -> list[tuple[Endpoint, str]]: + now = time.monotonic() + with self._lock: + start = self._rr + self._rr += 1 + n = len(self.endpoints) + order = [self.endpoints[(start + i) % n] for i in range(n)] if n else [] + out = [] + for ep in order: + if ep.down_until > now: + continue + m = self._serves(ep) + if m: + out.append((ep, m)) + return out + + def chat(self, system: str, user: str, **kw) -> tuple[str, dict]: + """One chat call on the next healthy endpoint; fails over to the others.""" + cands = self.candidates() + if not cands: + raise RuntimeError(f"no reachable Ollama endpoint serves {self.default_model!r} " + f"(tried {[e.host for e in self.endpoints]})") + last: Exception | None = None + for ep, model in cands: + think = self.default_think if ep.think == "default" else ep.think + ctx = ep.num_ctx if ep.num_ctx is not None else self.default_ctx + try: + content, stats = self.call(system, user, model=model, host=ep.host, num_ctx=ctx, think=think, **kw) + stats = {**stats, "host": ep.host, "model": model} + return content, stats + except (urllib.error.URLError, ConnectionError, TimeoutError, socket.timeout) as e: + last = e + except RuntimeError as e: # ollama_chat's HTTP errors: only 5xx is the host's fault + if "ollama 5" not in str(e): + raise + last = e + ep.down_until = time.monotonic() + self.down_seconds + raise RuntimeError(f"all Ollama endpoints failed; last error: {last}") + + +def from_settings(call=None) -> OllamaPool: + from youtube_claims.recap import settings + from youtube_claims.recap.extract import ollama_chat + return OllamaPool(parse_hosts(settings.OLLAMA_HOSTS), settings.OLLAMA_MODEL, settings.OLLAMA_NUM_CTX, + settings.OLLAMA_THINK, call=call or ollama_chat) diff --git a/youtube_claims/recap/nightly.cmd b/youtube_claims/recap/nightly.cmd new file mode 100644 index 0000000..9d1e101 --- /dev/null +++ b/youtube_claims/recap/nightly.cmd @@ -0,0 +1,16 @@ +@echo off +rem Nightly Ross recap loop -- run by Windows Task Scheduler on the owner's workstation. +rem Uses the durable venv data\ross\.venv (py -3.13 --system-site-packages + faster-whisper, yt-dlp): +rem py -3.13 -m venv --system-site-packages data\ross\.venv +rem data\ross\.venv\Scripts\python -m pip install faster-whisper yt-dlp +rem RECAP_* settings (RECAP_OLLAMA_HOSTS, RECAP_OLLAMA_MODEL, ...) can live in the repo .env. +rem Extra args pass through (e.g. --backfill, --report-only). Log: data\ross\logs\nightly_YYYY-MM-DD.log +setlocal +cd /d "%~dp0..\.." +set PYTHONIOENCODING=utf-8 +if not exist data\ross\logs mkdir data\ross\logs +for /f %%i in ('powershell -NoProfile -Command "Get-Date -Format yyyy-MM-dd"') do set RUNDATE=%%i +set PY=data\ross\.venv\Scripts\python.exe +if not exist %PY% set PY=py -3.13 +%PY% -m youtube_claims.recap.nightly %* >> "data\ross\logs\nightly_%RUNDATE%.log" 2>&1 +exit /b %ERRORLEVEL% diff --git a/youtube_claims/recap/nightly.py b/youtube_claims/recap/nightly.py new file mode 100644 index 0000000..0c17b12 --- /dev/null +++ b/youtube_claims/recap/nightly.py @@ -0,0 +1,197 @@ +"""The nightly pass (run on the owner's Windows workstation, ~20:30 ET weekdays): + + 1. list the channel's newest uploads (yt-dlp flat listing) and drop the ones already in + transcripts.recap_videos + 2. per new video: metadata -> audio -> faster-whisper -> bundle (meta + our scanner list + + transcript) -> extraction (the gold file if the pilot has one, else local Ollama with the + verbatim-quote check) -> join (tape-verified symbol + our paper/live view) -> Postgres + 3. write the cumulative report data/ross/reports/YYYY-MM-DD.md (+ .json for the next diff) + + python -m youtube_claims.recap.nightly # normal nightly run + python -m youtube_claims.recap.nightly --backfill # also ingest every local bundle not in the DB + python -m youtube_claims.recap.nightly --reprocess ID # force one video through again + python -m youtube_claims.recap.nightly --report-only + +Each video is independent: a failure is recorded on its recap_videos row and retried next +night (up to RECAP_MAX_ATTEMPTS), never aborting the run.""" + +from __future__ import annotations + +import argparse +import datetime as dt +import json +import os +import sys +import time +import traceback + +from youtube_claims.recap import settings + + +def log(*a): + print(dt.datetime.now().strftime("%H:%M:%S"), *a, flush=True) + + +def _stem(meta: dict) -> str: + return f"{meta['upload_date']}_{meta['id']}" + + +def extraction_for(stem: str, bundle: str, *, force_llm: bool = False) -> tuple[dict, str, dict | None]: + """(record, extractor, run_meta). A pilot gold file wins; else the cached local-model + output; else a fresh Ollama extraction (cached to data/ross/llm/).""" + from youtube_claims.recap.extract import extract_bundle + d = settings.dirs() + gold = d["extracted"] / f"{stem}.json" + if gold.exists() and not force_llm: + return json.loads(gold.read_text(encoding="utf-8")), "gold", None + cache = d["llm"] / f"{stem}.json" + if cache.exists() and not force_llm: + rec = json.loads(cache.read_text(encoding="utf-8")) + run = rec.pop("_run", None) + return rec, (run or {}).get("model") or settings.OLLAMA_MODEL, run + rec, run = extract_bundle(bundle, log=log) + cache.write_text(json.dumps({**rec, "_run": run}, indent=1), encoding="utf-8") + return rec, run["model"], run + + +def process(meta: dict, ourside, *, force_llm: bool = False) -> dict: + """One video end to end. Returns a small summary for the log.""" + from youtube_claims.recap import channel, store + from youtube_claims.recap.join import join_record + vid, stem = meta["id"], _stem(meta) + d = settings.dirs() + t0 = time.time() + scanner = channel.scanner_for(meta["upload_date"], ourside) + bfile = d["bundles"] / f"{stem}.txt" + if bfile.exists(): # built before (pilot or an earlier night): reuse as-is + bundle = bfile.read_text(encoding="utf-8") + src = channel.bundle_source(bundle) + else: + src, segs = channel.best_transcript(vid) + if not segs: + log(f"{vid}: downloading audio") + audio = channel.download_audio(vid) + log(f"{vid}: transcribing {audio.name}") + segs = channel.transcribe(vid, audio) + src = "whisper-small" + try: + audio.unlink() # the transcript is what we keep; audio is ~10 MB/video + except OSError: + pass + bundle = channel.make_bundle(meta, src, segs, scanner) + bfile.write_text(bundle, encoding="utf-8") + t_tx = time.time() - t0 + t1 = time.time() + rec, extractor, run = extraction_for(stem, bundle, force_llm=force_llm) + rec.setdefault("video_id", vid) + rec.setdefault("upload_date", meta["upload_date"]) + t_ex = time.time() - t1 + t2 = time.time() + trades, passed = join_record(rec, ourside, scanner) + t_join = time.time() - t2 + run = {**(run or {}), "timing_s": {"transcribe": round(t_tx), "extract": round(t_ex), "join": round(t_join)}} + store.upsert_video(rec, meta, extractor=extractor, transcript_source=src, run_meta=run, + trades=trades, passed=passed) + return {"kind": rec.get("video_kind"), "trades": len(trades), "passed": len(passed), "extractor": extractor, + "seconds": round(time.time() - t0)} + + +def _local_bundle_metas() -> list[dict]: + """Every pilot bundle on disk as a meta dict (for --backfill).""" + out = [] + for f in sorted(settings.dirs()["bundles"].glob("*.txt")): + head = f.read_text(encoding="utf-8").splitlines()[:3] + kv = dict(ln.split(": ", 1) for ln in head if ": " in ln) + mf = settings.dirs()["transcripts"] / f"{kv.get('video_id')}.meta.json" + dur = json.loads(mf.read_text(encoding="utf-8")).get("duration") if mf.exists() else None + out.append({"id": kv["video_id"], "upload_date": kv["upload_date"], "title": kv.get("title", ""), + "duration": dur}) + return out + + +def main(argv=None) -> int: + ap = argparse.ArgumentParser(description="Nightly Ross recap learning loop") + ap.add_argument("--backfill", action="store_true", help="ingest every local bundle not yet in the DB") + ap.add_argument("--reprocess", action="append", default=[], help="video id to force through again") + ap.add_argument("--force-llm", action="store_true", help="ignore gold/cached extractions for processed videos") + ap.add_argument("--report-only", action="store_true") + ap.add_argument("--no-list", action="store_true", help="skip the channel listing (offline)") + ap.add_argument("--max-new", type=int, default=settings.MAX_NEW_PER_RUN) + a = ap.parse_args(argv) + + from youtube_claims.recap import channel, report, store + from youtube_claims.recap.ourside import OurSide + d = settings.dirs() + store.apply_schema() + known = store.video_status() + max_attempts = int(os.getenv("RECAP_MAX_ATTEMPTS", "3")) + processed: list[str] = [] + failures: list[str] = [] + + if not a.report_only: + local = _local_bundle_metas() + local_ids = {m["id"] for m in local} + todo: list[dict] = [] + if not a.no_list: + try: + ups = channel.list_uploads() + log(f"channel listing: {len(ups)} newest uploads") + except Exception as e: # noqa: BLE001 -- still ingest/report what we have + log(f"channel listing FAILED: {e}") + failures.append(f"listing: {e}") + ups = [] + for u in ups: + k = known.get(u["id"]) + if k and (k["status"] in ("done", "skipped") or + (k["status"] == "failed" and (k.get("attempts") or 0) >= max_attempts)): + continue + if u["id"] in a.reprocess: + continue + try: + m = channel.video_meta(u["id"], u["title"], u.get("duration")) + except Exception as e: # noqa: BLE001 + log(f"{u['id']}: metadata failed: {e}") + failures.append(f"{u['id']} meta: {e}") + continue + if not m.get("upload_date") or m["upload_date"] < settings.MIN_UPLOAD_DATE: + continue + todo.append(m) + if a.backfill: + have = {m["id"] for m in todo} + todo += [m for m in local + if m["id"] not in have and (known.get(m["id"]) or {}).get("status") != "done"] + for vid in a.reprocess: + todo.append(channel.video_meta(vid)) + todo.sort(key=lambda m: m["upload_date"]) + new_from_channel = [m for m in todo if m["id"] not in local_ids] + if len(new_from_channel) > a.max_new: # politeness cap on downloads per night + drop = {m["id"] for m in new_from_channel[:-a.max_new]} + todo = [m for m in todo if m["id"] not in drop] + log(f"capped downloads at {a.max_new}; {len(drop)} older new videos wait for the next run") + log(f"{len(todo)} video(s) to process") + ours = OurSide() + try: + for m in todo: + try: + s = process(m, ours, force_llm=a.force_llm and m["id"] in a.reprocess) + processed.append(m["id"]) + log(f"{m['id']} {m['upload_date']} {m.get('title', '')[:50]!r}: {s}") + except Exception as e: # noqa: BLE001 -- one bad video never aborts the night + log(f"{m['id']}: FAILED {type(e).__name__}: {e}") + traceback.print_exc() + failures.append(f"{m['id']}: {type(e).__name__}: {str(e)[:200]}") + store.mark(m["id"], "failed", error=f"{type(e).__name__}: {e}", meta=m) + finally: + ours.close() + + ev = settings.DATA_DIR / "eval" / "_best.txt" + out = report.write_report(d["reports"], new_video_ids=processed, + eval_text=ev.read_text(encoding="utf-8") if ev.exists() else None) + log(f"report: {out} (processed {len(processed)}, failures {len(failures)})") + for f in failures: + log(" failure:", f) + return 1 if failures and not processed else 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/youtube_claims/recap/ourside.py b/youtube_claims/recap/ourside.py new file mode 100644 index 0000000..fdf97d3 --- /dev/null +++ b/youtube_claims/recap/ourside.py @@ -0,0 +1,112 @@ +"""What OUR system saw/did for a (date, symbol): discovery, state changes, gate verdicts, +signals, orders, closes, risk rules -- from BOTH stacks' event logs (paper DB :5433 and live +Tradier DB :5434) -- plus the day's SIP tape stats, float and news. Ported from the pilot's +ourside.py; read-only against the trading DBs. Event timestamps are ET-naive; minute_bars are +UTC-naive.""" + +from __future__ import annotations + +import re +from collections import Counter + +import psycopg2 + +from youtube_claims.recap import settings +from youtube_claims.recap.tape import BarSource, tape_stats + +_EVENT_TYPES = ["symbol_discovered", "symbol_state_changed", "signal_blocked", "entry_gate_evaluated", + "signal_ready", "order_submitted", "position_closed", "risk_rule_triggered"] + + +def norm_reason(m: str | None) -> str: + """Collapse numbers so '3.2x < 5x' and '4.1x < 5x' count as one reason.""" + return re.sub(r"\d+(\.\d+)?", "#", m or "")[:120] + + +def summarize_events(ev: list[tuple]) -> dict: + """ev: [(ts, event_type, message)] ordered by time -> the compact per-stack view.""" + hm = lambda t: t.strftime("%H:%M") # noqa: E731 + blocks = Counter(norm_reason(e[2].split("blocked:", 1)[-1]) for e in ev if e[1] == "signal_blocked") + gates = Counter(norm_reason(e[2].split("rejected:", 1)[-1]) for e in ev + if e[1] == "entry_gate_evaluated" and "rejected" in (e[2] or "")) + states: list[tuple] = [] + for e in ev: + if e[1] == "symbol_state_changed" and (not states or states[-1][1] != e[2][:160]): + states.append((hm(e[0]), (e[2] or "")[:160])) + pick = lambda typ, n, w: [(hm(e[0]), (e[2] or "")[:w]) for e in ev if e[1] == typ][:n] # noqa: E731 + return {"discovered": pick("symbol_discovered", 2, 200), "states": states[:8], + "top_blocks": blocks.most_common(3), "gate_rejects": gates.most_common(3), + "signals": pick("signal_ready", 3, 120), "orders": pick("order_submitted", 3, 100), + "closes": pick("position_closed", 3, 120), "risk": pick("risk_rule_triggered", 4, 140)} + + +class OurSide: + def __init__(self): + self.paper = psycopg2.connect(settings.database_url(), connect_timeout=10) + self.paper.autocommit = True + try: + self.live = psycopg2.connect(settings.live_database_url(), connect_timeout=10) + self.live.autocommit = True + except Exception as e: # noqa: BLE001 -- the paper view alone is still useful + print(f"[ourside] live DB unavailable: {e}") + self.live = None + self.bars = BarSource(settings.dirs()["cache"] / "bars", paper_con=self.paper) + + def close(self): + for c in (self.paper, self.live): + if c is not None: + c.close() + + def _q(self, con, sql, args=()): + with con.cursor() as cur: + cur.execute(sql, args) + return cur.fetchall() + + def events(self, con, d: str, sym: str) -> list[tuple]: + return self._q(con, "select timestamp, event_type, message from events where event_type = any(%s) " + "and timestamp between %s and %s and payload_json like %s order by timestamp", + (_EVENT_TYPES, f"{d} 04:00", f"{d} 20:00", f'%"symbol": "{sym}"%')) + + def prev_close(self, d: str, sym: str) -> float | None: + r = self._q(self.paper, "select close from daily_bars where symbol=%s and trade_date<%s " + "order by trade_date desc limit 1", (sym, d)) + return float(r[0][0]) if r and r[0][0] else None + + def view(self, d: str, sym: str) -> dict: + d, sym = str(d), sym.upper() + out: dict = {"date": d, "symbol": sym} + for name, con in (("paper", self.paper), ("live", self.live)): + out[name] = summarize_events(self.events(con, d, sym)) if con is not None else None + bars = self.bars.bars(sym, d) + out["tape"] = tape_stats(bars, self.prev_close(d, sym)) + local = self._q(self.paper, "select min(timestamp), count(*) from minute_bars where symbol=%s " + "and session_date=%s", (sym, d)) + if local and local[0][0]: + import datetime as dt + from zoneinfo import ZoneInfo + first = local[0][0].replace(tzinfo=dt.timezone.utc).astimezone(ZoneInfo("America/New_York")) + out["tape"]["our_first_bar"] = first.strftime("%H:%M") + out["tape"]["our_bar_count"] = local[0][1] if local else 0 + fl = self._q(self.paper, "select float_shares from float_cache where symbol=%s", (sym,)) + out["float"] = float(fl[0][0]) if fl and fl[0][0] else None + out["news"] = [r[0][:140] for r in self._q( + self.paper, "select headline from news_events where symbol=%s and published_at::date " + "between %s::date - 1 and %s::date order by published_at limit 3", (sym, d, d))] + return out + + def day_universe(self, d: str) -> list[tuple]: + """Symbols our scanner ranked that day: [(symbol, gap %, price)] (the bundle's list).""" + rows = self._q(self.paper, "select symbol, max(gap_pct), max(price) from scanner_snapshots " + "where snapshot_time::date=%s group by symbol " + "order by max(gap_pct) desc nulls last limit 40", (str(d),)) + if not rows: + rows = self._q(self.paper, "select symbol, gap_pct, null from historical_gappers " + "where session_date=%s order by rank limit 40", (str(d),)) + return [(s, round(float(g or 0) * (100 if g and abs(g) < 20 else 1), 1), + round(float(p), 4) if p else None) for s, g, p in rows] + + +def discovery_time(view: dict | None) -> str | None: + """Earliest HH:MM either stack discovered the symbol.""" + ts = [t for side in ("paper", "live") for t, _ in ((view or {}).get(side) or {}).get("discovered", [])] + return min(ts) if ts else None diff --git a/youtube_claims/recap/prompt.py b/youtube_claims/recap/prompt.py new file mode 100644 index 0000000..e00cba7 --- /dev/null +++ b/youtube_claims/recap/prompt.py @@ -0,0 +1,126 @@ +"""The extraction instructions (EXTRACT_SPEC.md, adapted for a local model) and the JSON +schema handed to Ollama's structured-output mode. The spec file itself lives in the +git-ignored data dir; this is the versioned copy the pipeline actually uses.""" + +from __future__ import annotations + +SETUPS = ["micro_pullback", "bull_flag", "first_pullback", "breakout_whole_dollar", + "break_of_premarket_high", "halt_resume", "dip_and_curl", "vwap_break", "abcd", + "red_to_green", "other"] +KINDS = ["daily_recap", "watchlist", "education", "challenge_summary", "other"] + +SYSTEM = """You extract a structured record from ONE transcript of Ross Cameron's (Warrior Trading) +day-trading video. Record what he SAID, not what you think is right. + +QUOTES ARE CHECKED BY A PROGRAM. Every `quote`, `setup_quote`, `catalyst_quote` and `market_quote` +must be copied word-for-word from the TRANSCRIPT (a contiguous run of 6-30 words, ASR noise +included; do not fix grammar or numbers). Anything whose quote is not found verbatim is deleted. +If you cannot quote it, leave the item out. + +The transcript is speech-to-text: tickers and prices are often misheard. Put the ticker AS HEARD in +`ticker_heard` and your best guess in `ticker_guess`, preferring a symbol from OUR SCANNER LIST when +it sounds alike and the price fits. Spoken prices: "550" usually means $5.50, "750" means $7.50, +"a dollar twenty" = 1.20, "four and a half" = 4.50. Prices are verified against the tape later, so +only give a price he actually said. + +What goes where: +- video_kind: daily_recap (he recaps his own trading of a session -- even if he mentions a challenge + or account milestone), watchlist (mainly stocks to watch for the next session), challenge_summary + (ONLY when the video summarises a multi-day/multi-week challenge as a whole), education (a lesson + or story not about one session), other. +- session_date: the trading day described (usually the upload date; the prior Friday for a weekend + watchlist; null if no specific day). +- trades: walk through the transcript IN ORDER and create ONE entry every time he says he bought, + got in, took a trade, added, or got back in ("I got in", "I bought", "I was in", "I added", + "I took a trade on", "I jumped in"). Several trades on the same stock are normal -- each re-entry + or add is its own entry (add_on=true). Trades he took but mentions only briefly still count, and + watchlist or challenge videos can contain trades too. Do NOT create trades for stocks he only + watched, showed on a chart, or talks about hypothetically. + setup: one of micro_pullback, bull_flag, first_pullback, breakout_whole_dollar, + break_of_premarket_high, halt_resume, dip_and_curl, vwap_break, abcd, red_to_green, other. + outcome: win / loss / scratch as he describes it, else null. pnl/shares only if stated. +- passed_on: EVERY other stock he names that he did NOT trade -- top gainers he skipped, stocks on + his watchlist, names he says he missed or avoided -- with the reason (or what he said about it) and + a quote naming or describing it. Usually 3-10 per video. +- ticker_guess: when the heard name is close to a symbol in OUR SCANNER LIST (sound-alike, spelled + letters, or the price he mentions fits), use that scanner symbol. +- rules_stated: general principles he states (risk, sizing, stock selection, when to stop, + statistics about his accuracy or P&L), each with a quote. +- day.market_temperature: hot (lots of big clean movers, strong momentum), cold (slow, few good + setups, choppy, tough month), mixed, or null if he gives no read. market_quote: the sentence where + he characterises the market. +- day.net_pnl: the day's $ result as stated (negative for red); green = true if the day is positive. + +Keep paraphrases (reason, rule, lesson, read) short. Skip sponsor segments, course promotion and +filler. Return ONLY the JSON object.""" + +USER_TMPL = """{part_note}Upload metadata and the day's context follow, then the transcript. + +{bundle}""" + +PART_NOTE = ("This is PART {i} of {n} of one transcript (parts overlap slightly). Extract only what " + "is said in this part; fill `day` fields only if this part states them.\n\n") + + +def _nstr(): + return {"type": ["string", "null"]} + + +def _nnum(): + return {"type": ["number", "null"]} + + +def _quoted(key: str) -> dict: + return {"type": "array", "items": {"type": "object", "properties": { + key: {"type": "string"}, "quote": {"type": "string"}}, "required": [key, "quote"]}} + + +SCHEMA = { + "type": "object", + "properties": { + "video_kind": {"type": "string", "enum": KINDS}, + "session_date": _nstr(), + "day": {"type": "object", "properties": { + "net_pnl": _nnum(), "green": {"type": ["boolean", "null"]}, + "market_temperature": {"type": ["string", "null"], "enum": ["hot", "cold", "mixed", None]}, + "market_quote": _nstr(), "max_loss_hit": {"type": ["boolean", "null"]}, + "daily_goal_comment": _nstr(), "sizing_comment": _nstr()}, + "required": ["net_pnl", "green", "market_temperature", "market_quote"]}, + "trades": {"type": "array", "items": {"type": "object", "properties": { + "ticker_heard": {"type": "string"}, "ticker_guess": {"type": "string"}, + "entry_time_hint": _nstr(), + "setup": {"type": "string", "enum": SETUPS}, "setup_quote": {"type": "string"}, + "entry_price": _nnum(), "exit_price": _nnum(), "stated_high": _nnum(), "stated_low": _nnum(), + "shares": _nnum(), "pnl": _nnum(), + "outcome": {"type": ["string", "null"], "enum": ["win", "loss", "scratch", None]}, + "add_on": {"type": "boolean"}, + "stock_context": {"type": "object", "properties": { + "catalyst": _nstr(), "catalyst_quote": _nstr(), "float": _nstr(), + "pct_up_at_entry": _nstr(), "leading_gainer_rank": _nnum(), + "other": {"type": "array", "items": {"type": "string"}}}}, + "why_entered": _quoted("reason"), "tape_reads": _quoted("read"), + "why_exited": _quoted("reason"), "mistake_or_lesson": _quoted("lesson")}, + "required": ["ticker_heard", "ticker_guess", "setup", "setup_quote", "entry_price", + "exit_price", "outcome", "why_entered"]}}, + "passed_on": {"type": "array", "items": {"type": "object", "properties": { + "ticker_heard": {"type": "string"}, "ticker_guess": {"type": "string"}, + "reason": {"type": "string"}, "quote": {"type": "string"}}, + "required": ["ticker_heard", "ticker_guess", "reason", "quote"]}}, + "rules_stated": _quoted("rule"), + }, + "required": ["video_kind", "session_date", "day", "trades", "passed_on", "rules_stated"], +} + + +_TASK_A = ("TASK FOR THIS CALL: your main job is `trades` -- list EVERY position he took, in order, " + "with its quotes and prices. Also fill video_kind, session_date and day (net P&L, " + "market_temperature). Passed-on stocks and rules are extracted in a separate call.\n\n") +_TASK_B = ("TASK FOR THIS CALL: fill passed_on (every stock he names but did NOT trade) and " + "rules_stated (every general principle, statistic or lesson he states -- usually 5-12). " + "Trades are extracted in a separate call.\n\n") +SCHEMA_A = {"type": "object", "properties": {k: SCHEMA["properties"][k] for k in + ("video_kind", "session_date", "day", "trades")}, + "required": ["video_kind", "session_date", "day", "trades"]} +SCHEMA_B = {"type": "object", "properties": {k: SCHEMA["properties"][k] for k in ("passed_on", "rules_stated")}, + "required": ["passed_on", "rules_stated"]} +SPLIT_TASKS = [(_TASK_A, SCHEMA_A), (_TASK_B, SCHEMA_B)] diff --git a/youtube_claims/recap/quotes.py b/youtube_claims/recap/quotes.py new file mode 100644 index 0000000..42b051f --- /dev/null +++ b/youtube_claims/recap/quotes.py @@ -0,0 +1,94 @@ +"""Anti-hallucination: every quote the LLM returns must be findable in the transcript. + +Extends youtube_claims.extraction._quote_supported (whitespace-normalized 40-char prefix match, +which accepts a quote whose tail is invented) with a full-length, punctuation-insensitive token +match plus a small fuzzy tolerance for a mis-copied word, +and SNAPS an accepted quote to the transcript's own words -- so what we store is verbatim +source text, never the model's paraphrase of it. A quote that can't be located is rejected +and the caller drops the item it supported.""" + +from __future__ import annotations + +import re +from difflib import SequenceMatcher + +from youtube_claims.extraction import _norm + +_TOKEN = re.compile(r"[a-z0-9]+(?:[.,'][a-z0-9]+)*") + + +def _tok(s: str) -> list[str]: + """Lowercased word tokens; '$5.50' -> '5.50', "don't" -> "don't", 'up-to' -> 'up','to'.""" + return _TOKEN.findall((s or "").lower()) + + +class Transcript: + """A transcript indexed for quote lookup. Keeps the original words so a matched quote can + be returned exactly as transcribed (ASR noise included).""" + + def __init__(self, text: str): + self.text = text or "" + self.ntext = _norm(self.text) + self.words = self.text.split() + self.wtoks = [_tok(w) for w in self.words] # tokens per original word + self.toks: list[str] = [] + self.tok_word: list[int] = [] # token index -> word index + for wi, ts in enumerate(self.wtoks): + for t in ts: + self.toks.append(t) + self.tok_word.append(wi) + self.joined = " " + " ".join(self.toks) + " " + # number tokens also without thousands separators / leading '$' for price lookups + self.token_set = set(self.toks) | {t.replace(",", "") for t in self.toks} + self._first: dict[str, list[int]] = {} + for i, t in enumerate(self.toks): + self._first.setdefault(t, []).append(i) + + def _span(self, i: int, n: int) -> str: + a, b = self.tok_word[i], self.tok_word[min(i + n, len(self.toks)) - 1] + return " ".join(self.words[a:b + 1]) + + def find(self, quote: str, min_ratio: float = 0.9) -> str | None: + """The transcript text matching `quote`, or None if it isn't there. + Exact (token-level, punctuation-blind) first; then a fuzzy window of the same length + whose tokens agree >= min_ratio (tolerates one mis-copied word in ten).""" + q = _tok(quote) + if len(q) < 3: # too short to be evidence of anything + return None + needle = " " + " ".join(q) + " " + pos = self.joined.find(needle) + if pos >= 0: + start = self.joined[:pos + 1].count(" ") - 1 + return self._span(start, len(q)) + if len(q) < 6: + return None + n = len(q) + cands: set[int] = set() + for k in range(min(4, n)): # windows anchored on any of the + for i in self._first.get(q[k], ()): # quote's first four tokens + if i - k >= 0: + cands.add(i - k) + best, best_i = 0.0, None + for i in sorted(cands): + win = self.toks[i:i + n] + if len(win) < n * 0.8: + continue + r = SequenceMatcher(None, q, win, autojunk=False).ratio() + if r > best: + best, best_i = r, i + if best_i is None or best < min_ratio: + return None + # a dropped/added word shifts the true end: pick the window length that fits best + best_n = max(range(max(1, n - 2), n + 3), + key=lambda m: (SequenceMatcher(None, q, self.toks[best_i:best_i + m], autojunk=False).ratio(), -abs(m - n))) + return self._span(best_i, best_n) + + def supports(self, quote: str) -> bool: + return self.find(quote) is not None + + +def transcript_text_from_bundle(bundle: str) -> str: + """The transcript body of a bundle, without the [mm:ss] prefixes (segments joined by + spaces so a quote spanning two segments still matches).""" + body = bundle.split("TRANSCRIPT (mm:ss text):", 1)[-1] + return " ".join(re.sub(r"^\[\d+:\d+\]\s?", "", ln) for ln in body.splitlines() if ln.strip()) diff --git a/youtube_claims/recap/report.py b/youtube_claims/recap/report.py new file mode 100644 index 0000000..76f2507 --- /dev/null +++ b/youtube_claims/recap/report.py @@ -0,0 +1,231 @@ +"""Cumulative gap report: Ross's trades vs what our system saw/did (the pilot's report.py +sections) plus "what changed since last run". compute()/render()/diff_metrics() are pure; +write_report() loads from Postgres and writes data/ross/reports/YYYY-MM-DD.{md,json}.""" + +from __future__ import annotations + +import datetime as dt +import json +import re +import statistics +from collections import Counter, defaultdict +from pathlib import Path + +GOOD = ("price_verified", "price_verified_loose", "price_matched_scanner") +BUCKETS = ["pre-7:00", "7:00-9:30", "9:30-9:45", "9:45+"] + + +def _f(x): + try: + return float(x) if x is not None else None + except (TypeError, ValueError): + return None + + +def _bucket(hhmm: str | None) -> str | None: + if not hhmm: + return None + return ("pre-7:00" if hhmm < "07:00" else "7:00-9:30" if hhmm < "09:30" + else "9:30-9:45" if hhmm < "09:45" else "9:45+") + + +def _disc(view: dict | None) -> str | None: + ts = [t for side in ("paper", "live") for t, *_ in ((view or {}).get(side) or {}).get("discovered", [])] + return min(ts) if ts else None + + +def _describe(xs: list[float]) -> dict: + if not xs: + return {"count": 0} + xs = sorted(xs) + q = statistics.quantiles(xs, n=4) if len(xs) > 1 else [xs[0]] * 3 + return {"count": len(xs), "mean": round(statistics.mean(xs), 2), "min": round(xs[0], 2), + "p25": round(q[0], 2), "median": round(q[1], 2), "p75": round(q[2], 2), "max": round(xs[-1], 2)} + + +def trade_facts(t: dict) -> dict: + """The per-trade columns the pilot report aggregated (also the CSV row).""" + o = t.get("our_view") or {} + tp = o.get("tape") or {} + prev, e = _f(tp.get("prev_close")), _f(t.get("entry_price")) + how = (t.get("resolve") or "").split(":")[0] + et = t.get("entry_time_tape") if how in GOOD else None + disc = _disc(o) + states = [s for side in ("paper", "live") for _, s in ((o.get(side) or {}).get("states") or [])] + blocks = [b for side in ("paper", "live") for b, _ in ((o.get(side) or {}).get("top_blocks") or [])] + sc = t.get("stock_context") or {} + return { + "video": t.get("video_id"), "date": str(t.get("session_date") or o.get("date") or ""), + "sym": t.get("symbol"), "resolve": how, "setup": t.get("setup"), "outcome": t.get("outcome"), + "pnl": _f(t.get("pnl")), "entry": e, "entry_t": et, "entry_bucket": _bucket(et), + "gap_at_entry": (e / prev - 1) if (e and prev) else None, + "catalyst": sc.get("catalyst"), "discovered": disc, + "discovered_before_entry": bool(disc and et and disc <= et), + "ours_traded": any(((o.get(s) or {}).get("orders")) for s in ("paper", "live")), + "out_of_population": next((s for s in states if "out_of_population" in s), None), + "top_block": blocks[0] if blocks else None, + } + + +def compute(videos: list[dict], trades: list[dict], passed: list[dict]) -> dict: + """All report numbers as a JSON-able dict (the diff compares two of these).""" + T = [trade_facts(t) for t in trades] + V = [r for r in T if r["resolve"] in GOOD] + m: dict = { + "videos": len(videos), "trades": len(T), "passed_on": len(passed), + "video_ids": sorted(v["video_id"] for v in videos), + "video_kinds": dict(Counter(v.get("video_kind") for v in videos)), + "extractors": dict(Counter(v.get("extractor") for v in videos)), + "resolution": dict(Counter(r["resolve"] for r in T).most_common()), + "verified": len(V), + "entry_buckets": {b: sum(r["entry_bucket"] == b for r in V) for b in BUCKETS + [None]}, + } + m["entry_buckets"]["unlocated"] = m["entry_buckets"].pop(None) + wl: dict = defaultdict(Counter) + pn: dict = defaultdict(list) + for r in V: + if r["entry_bucket"]: + wl[r["entry_bucket"]][r["outcome"] or "unstated"] += 1 + if r["pnl"] is not None: + pn[r["entry_bucket"]].append(r["pnl"]) + m["outcome_by_bucket"] = {b: dict(wl[b]) for b in BUCKETS if b in wl} + m["pnl_by_bucket"] = {b: {"count": len(pn[b]), "sum": round(sum(pn[b])), "median": round(statistics.median(pn[b]))} + for b in BUCKETS if pn.get(b)} + m["setups"] = dict(Counter(r["setup"] for r in V).most_common()) + m["setups_all"] = dict(Counter(r["setup"] for r in T).most_common()) + m["coverage"] = {"verified": len(V), "discovered": sum(r["discovered"] is not None for r in V), + "discovered_before_entry": sum(r["discovered_before_entry"] for r in V), + "we_ordered": sum(r["ours_traded"] for r in V), + "out_of_population": sum(r["out_of_population"] is not None for r in V)} + prices = [r["entry"] for r in V if r["entry"] is not None] + m["price_at_entry"] = _describe(prices) + m["gap_at_entry"] = _describe([r["gap_at_entry"] for r in V if r["gap_at_entry"] is not None]) + m["entries_under_2"] = sum(p < 2 for p in prices) + m["entries_over_20"] = sum(p > 20 for p in prices) + m["oop_reasons"] = Counter(re.sub(r"[\d.]+", "#", r["out_of_population"].split("out_of_population", 1)[1])[:80] + for r in V if r["out_of_population"]).most_common(8) + m["top_blocks"] = Counter(r["top_block"] for r in V if r["top_block"]).most_common(6) + m["catalysts"] = Counter((r["catalyst"] if isinstance(r["catalyst"], str) else "none/unstated").lower()[:30] + for r in V).most_common(10) + pr = [] + for p in passed: + tp = ((p.get("our_view") or {}).get("tape") or {}) + if p.get("symbol") and tp.get("open"): + pr.append((tp.get("open_to_hod_pct"), (tp["close"] / tp["open"] - 1) if tp.get("close") else None)) + oh = [a for a, _ in pr if a is not None] + oc = [b for _, b in pr if b is not None] + m["passed_tape"] = {"with_tape": len(pr), "median_open_to_hod": round(statistics.median(oh), 4) if oh else None, + "median_open_to_close": round(statistics.median(oc), 4) if oc else None} + temps = {} + for v in videos: + if v.get("session_date") and v.get("market_temperature"): + temps[str(v["session_date"])] = v["market_temperature"] + m["temperature_by_day"] = dict(sorted(temps.items())) + m["temperature"] = dict(Counter(temps.values())) + m["trade_keys"] = sorted(f"{r['video']}|{r['sym'] or '?'}|{r['setup']}|{r['entry']}" for r in T) + return m + + +def diff_metrics(prev: dict | None, cur: dict) -> list[str]: + """Human lines describing what changed between two compute() results.""" + if not prev: + return ["First report -- no previous run to compare with."] + out = [] + new_v = sorted(set(cur.get("video_ids", [])) - set(prev.get("video_ids", []))) + gone_v = sorted(set(prev.get("video_ids", [])) - set(cur.get("video_ids", []))) + if new_v: + out.append(f"New videos ({len(new_v)}): " + ", ".join(new_v)) + if gone_v: + out.append(f"Videos no longer in the set ({len(gone_v)}): " + ", ".join(gone_v)) + for k in ("videos", "trades", "passed_on", "verified"): + if prev.get(k) != cur.get(k): + out.append(f"{k}: {prev.get(k)} -> {cur.get(k)} ({cur.get(k, 0) - (prev.get(k) or 0):+d})") + for group in ("coverage", "entry_buckets", "temperature", "setups", "resolution"): + a, b = prev.get(group) or {}, cur.get(group) or {} + for k in sorted(set(a) | set(b), key=str): + if a.get(k, 0) != b.get(k, 0): + out.append(f"{group}.{k}: {a.get(k, 0)} -> {b.get(k, 0)}") + pa, pb = (prev.get("price_at_entry") or {}).get("median"), (cur.get("price_at_entry") or {}).get("median") + if pa != pb: + out.append(f"median entry price: {pa} -> {pb}") + new_t = sorted(set(cur.get("trade_keys", [])) - set(prev.get("trade_keys", []))) + if new_t: + out.append(f"New trades ({len(new_t)}): " + "; ".join(k.split("|", 1)[1] for k in new_t[:20])) + return out or ["No change since the previous run."] + + +def _table(rows: list[list], head: list[str]) -> str: + out = ["| " + " | ".join(head) + " |", "|" + "---|" * len(head)] + out += ["| " + " | ".join("" if c is None else str(c) for c in r) + " |" for r in rows] + return "\n".join(out) + + +def render(m: dict, changes: list[str], *, today: str, new_trades: list[dict] | None = None, + eval_text: str | None = None) -> str: + pct = lambda x: "-" if x is None else f"{x:.1%}" # noqa: E731 + cov = m["coverage"] + L = [f"# Ross recap vs our bot -- {today}", "", + f"Videos {m['videos']} ({', '.join(f'{k} {v}' for k, v in m['video_kinds'].items())}), " + f"trades {m['trades']}, passed-on {m['passed_on']}. Extractors: " + + ", ".join(f"{k} {v}" for k, v in m["extractors"].items()) + ".", "", + "## What changed since last run", ""] + [f"- {c}" for c in changes] + [""] + if new_trades: + L += ["## New trades this run", "", _table( + [[t.get("session_date"), t.get("symbol") or f"?{t.get('ticker_guess')}", t.get("resolve"), t.get("setup"), + t.get("entry_price"), t.get("outcome"), t.get("entry_time_tape"), _disc(t.get("our_view")), + "yes" if any(((t.get("our_view") or {}).get(s) or {}).get("orders") for s in ("paper", "live")) else ""] + for t in new_trades], + ["date", "symbol", "resolve", "setup", "entry", "outcome", "entry@tape", "we discovered", "we ordered"]), ""] + L += ["## Ticker resolution (trades)", "", + ", ".join(f"{k} {v}" for k, v in m["resolution"].items()) or "-", "", + f"Price-verified trades: {m['verified']} of {m['trades']}", "", + "## When he entered (verified, entry located on tape)", "", + _table([[b, m["entry_buckets"].get(b, 0)] for b in BUCKETS + ["unlocated"]], ["window", "trades"]), "", + "## Win/loss by entry window (self-reported)", ""] + outs = sorted({o for d in m["outcome_by_bucket"].values() for o in d}) + L += [_table([[b] + [m["outcome_by_bucket"][b].get(o, 0) for o in outs] for b in m["outcome_by_bucket"]], + ["window"] + outs) if outs else "-", "", + "## Stated $ P&L by entry window", "", + _table([[b, v["count"], v["sum"], v["median"]] for b, v in m["pnl_by_bucket"].items()], + ["window", "count", "sum", "median"]) if m["pnl_by_bucket"] else "-", "", + "## Setups (verified trades)", "", ", ".join(f"{k} {v}" for k, v in m["setups"].items()) or "-", "", + "## Our coverage of his verified trades", "", + f"- discovered by our scanner at all: {cov['discovered']} / {cov['verified']}", + f"- discovered BEFORE his entry: {cov['discovered_before_entry']}", + f"- we placed an order that day: {cov['we_ordered']}", + f"- rejected as out_of_population: {cov['out_of_population']}", "", + "## Price and gap at his entry", "", + f"- price: {m['price_at_entry']}", + f"- gap vs prior close: {m['gap_at_entry']}", + f"- entries under $2: {m['entries_under_2']}, over $20: {m['entries_over_20']}", "", + "## out_of_population reasons", ""] + [f"- {k} ({v})" for k, v in m["oop_reasons"]] + [ + "", "## Our top block reasons on his names", ""] + [f"- {k} ({v})" for k, v in m["top_blocks"]] + [ + "", "## Catalysts he cited", "", ", ".join(f"{k} {v}" for k, v in m["catalysts"]) or "-", "", + "## Passed-on names: what happened after he passed", "", + f"{m['passed_tape']['with_tape']} with tape; median open->HOD {pct(m['passed_tape']['median_open_to_hod'])}, " + f"open->close {pct(m['passed_tape']['median_open_to_close'])}", "", + "## Market temperature", "", ", ".join(f"{k} {v}" for k, v in m["temperature"].items()) or "-", "", + ", ".join(f"{d} {t}" for d, t in list(m["temperature_by_day"].items())[-15:]), ""] + if eval_text: + L += ["## Extraction quality (local model vs gold set)", "", "```", eval_text, "```", ""] + return "\n".join(L) + + +def previous_metrics(reports_dir: Path, today: str) -> dict | None: + prior = sorted(p for p in reports_dir.glob("*.json") if p.stem < today) + return json.loads(prior[-1].read_text(encoding="utf-8")) if prior else None + + +def write_report(reports_dir: Path, *, new_video_ids: list[str] | None = None, eval_text: str | None = None, + today: str | None = None) -> Path: + from youtube_claims.recap import store + today = today or dt.date.today().isoformat() + videos, trades, passed = store.load_rows() + m = compute(videos, trades, passed) + changes = diff_metrics(previous_metrics(reports_dir, today), m) + new_trades = [t for t in trades if t["video_id"] in set(new_video_ids or [])] + md = render(m, changes, today=today, new_trades=new_trades, eval_text=eval_text) + out = reports_dir / f"{today}.md" + out.write_text(md, encoding="utf-8") + (reports_dir / f"{today}.json").write_text(json.dumps(m, default=str, indent=1), encoding="utf-8") + return out diff --git a/youtube_claims/recap/schema.sql b/youtube_claims/recap/schema.sql new file mode 100644 index 0000000..8ffd48c --- /dev/null +++ b/youtube_claims/recap/schema.sql @@ -0,0 +1,94 @@ +-- Ross Cameron recap learning loop (youtube_claims.recap). Idempotent; {SCHEMA} is substituted +-- at apply time (default: transcripts). Every child table is keyed by video_id and rewritten +-- whole on re-ingest, so re-running a video never duplicates rows. + +CREATE SCHEMA IF NOT EXISTS {SCHEMA}; + +-- one row per recap video the loop has processed (or tried to) +CREATE TABLE IF NOT EXISTS {SCHEMA}.recap_videos ( + video_id TEXT PRIMARY KEY, + upload_date DATE, + session_date DATE, + title TEXT, + duration_s INT, + video_kind TEXT, -- daily_recap|watchlist|education|challenge_summary|other + transcript_source TEXT, -- whisper-small|whisper-medium|captions + extractor TEXT, -- 'gold' (strong-model pilot file) or the Ollama model name + status TEXT NOT NULL DEFAULT 'pending', -- pending|done|failed|skipped + last_error TEXT, + attempts INT NOT NULL DEFAULT 0, -- failed attempts (nightly retries up to a cap) + run_meta JSONB, -- chunks, seconds, quote-drop counts + record JSONB, -- the full extraction record (EXTRACT_SPEC) + first_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(), + processed_at TIMESTAMPTZ, + updated_at TIMESTAMPTZ NOT NULL DEFAULT now() +); +CREATE INDEX IF NOT EXISTS recap_videos_session_idx ON {SCHEMA}.recap_videos(session_date); + +-- the day as he described it (one per video; several videos can describe one session) +CREATE TABLE IF NOT EXISTS {SCHEMA}.recap_day_summary ( + video_id TEXT PRIMARY KEY REFERENCES {SCHEMA}.recap_videos(video_id) ON DELETE CASCADE, + session_date DATE, + net_pnl NUMERIC, + green BOOLEAN, + market_temperature TEXT, -- hot|cold|mixed + market_quote TEXT, + max_loss_hit BOOLEAN, + daily_goal_comment TEXT, + sizing_comment TEXT +); + +-- trades he described, joined to the tape and to what OUR bot saw +CREATE TABLE IF NOT EXISTS {SCHEMA}.recap_trades ( + video_id TEXT NOT NULL REFERENCES {SCHEMA}.recap_videos(video_id) ON DELETE CASCADE, + seq INT NOT NULL, + session_date DATE, + ticker_heard TEXT, + ticker_guess TEXT, + symbol TEXT, -- after price verification (NULL if unresolved) + resolve TEXT, -- price_verified|price_matched_scanner|ambiguous:..|no_price_match|unverified + setup TEXT, + setup_quote TEXT, + entry_price NUMERIC, + exit_price NUMERIC, + stated_high NUMERIC, + stated_low NUMERIC, + shares NUMERIC, + pnl NUMERIC, + outcome TEXT, + add_on BOOLEAN, + entry_time_hint TEXT, + entry_time_tape TEXT, -- HH:MM ET the tape first traded his entry price + stock_context JSONB, + why_entered JSONB, + tape_reads JSONB, + why_exited JSONB, + mistake_or_lesson JSONB, + our_view JSONB, -- discovery/gates/orders on paper+live, tape stats, float, news + PRIMARY KEY (video_id, seq) +); +CREATE INDEX IF NOT EXISTS recap_trades_symbol_idx ON {SCHEMA}.recap_trades(symbol, session_date); + +-- stocks he looked at and did NOT trade +CREATE TABLE IF NOT EXISTS {SCHEMA}.recap_passed_on ( + video_id TEXT NOT NULL REFERENCES {SCHEMA}.recap_videos(video_id) ON DELETE CASCADE, + seq INT NOT NULL, + session_date DATE, + ticker_heard TEXT, + ticker_guess TEXT, + symbol TEXT, + resolve TEXT, + reason TEXT, + quote TEXT, + our_view JSONB, + PRIMARY KEY (video_id, seq) +); + +-- general principles he stated, verbatim-quoted +CREATE TABLE IF NOT EXISTS {SCHEMA}.recap_rules ( + video_id TEXT NOT NULL REFERENCES {SCHEMA}.recap_videos(video_id) ON DELETE CASCADE, + seq INT NOT NULL, + rule TEXT, + quote TEXT NOT NULL, + PRIMARY KEY (video_id, seq) +); diff --git a/youtube_claims/recap/settings.py b/youtube_claims/recap/settings.py new file mode 100644 index 0000000..a93a459 --- /dev/null +++ b/youtube_claims/recap/settings.py @@ -0,0 +1,75 @@ +"""Recap-loop settings. Env-driven, deliberately decoupled from the trading bot's OLLAMA_* +vars (those point the bot's news enrichment elsewhere and are rewritten by config_overrides). +Secrets (DATABASE_URL, Alpaca keys) come only from the repo .env.""" + +from __future__ import annotations + +import os +from pathlib import Path + +# RECAP_ENV_FILE lets a worktree/other checkout borrow the main checkout's .env. +if os.getenv("RECAP_ENV_FILE"): + try: + from dotenv import load_dotenv + load_dotenv(os.environ["RECAP_ENV_FILE"]) + except Exception: # noqa: BLE001 + pass + +from youtube_claims import config as _yc # noqa: E402 -- loads the repo .env (DATABASE_URL, Alpaca) + +ROOT = Path(__file__).resolve().parents[2] + +# Data lives in the git-ignored data dir of the checkout that runs the loop. +DATA_DIR = Path(os.getenv("RECAP_DATA_DIR") or ROOT / "data" / "ross") + +CHANNEL_URL = os.getenv("RECAP_CHANNEL_URL", "https://www.youtube.com/@DaytradeWarrior/videos") +CHANNEL_NAME = "Ross Cameron - Warrior Trading" +# Only videos uploaded on/after this date are processed (the pilot's window start). +MIN_UPLOAD_DATE = os.getenv("RECAP_MIN_UPLOAD_DATE", "20260601") +LIST_LIMIT = int(os.getenv("RECAP_LIST_LIMIT", "40")) # newest N uploads checked per run +MAX_NEW_PER_RUN = int(os.getenv("RECAP_MAX_NEW_PER_RUN", "8")) + +# Extraction LLM: the owner's LOCAL Ollama on this workstation (gpt-oss:20b on its GPU). It was +# the more precise of the two available models on the 36-video gold set (fewer invented prices); +# the server's 4B tools:latest had slightly higher recall -- see evaluate.py / the commit message. +OLLAMA_HOST = os.getenv("RECAP_OLLAMA_HOST", "http://127.0.0.1:11434") +OLLAMA_MODEL = os.getenv("RECAP_OLLAMA_MODEL", "gpt-oss:20b") +# All endpoints to spread/fail over across (comma list; "host=model#ctx=N#think=V" pins per host). +# Defaults to the single RECAP_OLLAMA_HOST. See youtube_claims/recap/llm.py. +OLLAMA_HOSTS = os.getenv("RECAP_OLLAMA_HOSTS") or OLLAMA_HOST +OLLAMA_NUM_PREDICT = int(os.getenv("RECAP_OLLAMA_NUM_PREDICT", "12000")) +OLLAMA_NUM_CTX = int(os.getenv("RECAP_OLLAMA_NUM_CTX", "24576")) or None # 0 -> model default +# gpt-oss reasoning effort ("low"; "medium" ran past the token budget and took ~10 min/video). +# "false" for non-reasoning models. +OLLAMA_THINK = {"false": False, "true": True, "": None}.get(os.getenv("RECAP_OLLAMA_THINK", "low").lower(), + os.getenv("RECAP_OLLAMA_THINK", "low").lower()) +OLLAMA_TIMEOUT = int(os.getenv("RECAP_OLLAMA_TIMEOUT", "1800")) +# Two focused calls per chunk (trades+day, then passed-on+rules) instead of one. +SPLIT_CALLS = os.getenv("RECAP_SPLIT_CALLS", "1").lower() in ("1", "true", "yes") +# Transcripts longer than this (chars) are split into overlapping chunks and merged. +CHUNK_CHARS = int(os.getenv("RECAP_CHUNK_CHARS", "24000")) + +# Whisper (CPU int8; the pilot's settings -- ~7 min per 15-min video with 6 threads under load). +WHISPER_MODEL = os.getenv("RECAP_WHISPER_MODEL", "small.en") +WHISPER_THREADS = int(os.getenv("RECAP_WHISPER_THREADS", "8")) # more threads was slower here +WHISPER_PROMPT = ("Ross Cameron day trading recap. Tickers, prices, float, relative volume, level 2, " + "micro pullback, bull flag, halt, VWAP, premarket high, short squeeze.") + +PG_SCHEMA = _yc.PG_SCHEMA + + +def database_url() -> str: + return _yc.database_url() + + +def live_database_url() -> str: + """The live (Tradier) stack's DB: same server/credentials, port 5434.""" + return os.getenv("RECAP_LIVE_DATABASE_URL") or database_url().replace(":5433", ":5434") + + +def dirs() -> dict[str, Path]: + d = {k: DATA_DIR / k for k in ("transcripts", "audio", "bundles", "extracted", "llm", + "scanner", "reports", "cache")} + for p in d.values(): + p.mkdir(parents=True, exist_ok=True) + return d diff --git a/youtube_claims/recap/store.py b/youtube_claims/recap/store.py new file mode 100644 index 0000000..454b71c --- /dev/null +++ b/youtube_claims/recap/store.py @@ -0,0 +1,139 @@ +"""Postgres persistence for the recap loop (transcripts schema on the paper/research DB). +Idempotent: DDL is IF NOT EXISTS; ingest upserts the video row and rewrites its child rows in +one transaction, keyed by video_id.""" + +from __future__ import annotations + +import json +from contextlib import contextmanager +from datetime import datetime, timezone +from pathlib import Path + +import psycopg2 +import psycopg2.extras + +from youtube_claims.recap import settings + +S = settings.PG_SCHEMA + + +@contextmanager +def connect(): + conn = psycopg2.connect(settings.database_url(), connect_timeout=10) + try: + yield conn + finally: + conn.close() + + +def apply_schema() -> None: + ddl = (Path(__file__).parent / "schema.sql").read_text(encoding="utf-8").replace("{SCHEMA}", S) + with connect() as conn: + with conn.cursor() as cur: + cur.execute(ddl) + conn.commit() + + +def video_status() -> dict[str, dict]: + with connect() as conn, conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(f"SELECT video_id, status, extractor, upload_date, attempts FROM {S}.recap_videos") + return {r["video_id"]: dict(r) for r in cur.fetchall()} + + +def _date(s): + if not s: + return None + s = str(s) + return f"{s[:4]}-{s[4:6]}-{s[6:]}" if len(s) == 8 and s.isdigit() else s + + +def _j(x): + return json.dumps(x, default=str) if x is not None else None + + +def mark(video_id: str, status: str, *, error: str | None = None, meta: dict | None = None) -> None: + """Record a video's status without touching its extraction (used for failures/skips).""" + meta = meta or {} + with connect() as conn, conn.cursor() as cur: + cur.execute( + f"INSERT INTO {S}.recap_videos (video_id, upload_date, title, duration_s, status, last_error, attempts) " + f"VALUES (%s,%s,%s,%s,%s,%s,%s) ON CONFLICT (video_id) DO UPDATE SET status=EXCLUDED.status, " + f"last_error=EXCLUDED.last_error, " + f"attempts={S}.recap_videos.attempts + (CASE WHEN EXCLUDED.status='failed' THEN 1 ELSE 0 END), title=COALESCE(EXCLUDED.title, {S}.recap_videos.title), " + f"upload_date=COALESCE(EXCLUDED.upload_date, {S}.recap_videos.upload_date), updated_at=now()", + (video_id, _date(meta.get("upload_date")), meta.get("title"), + int(meta["duration"]) if str(meta.get("duration") or "").isdigit() else None, status, + (error or "")[:2000] or None, 1 if status == "failed" else 0)) + conn.commit() + + +def upsert_video(rec: dict, meta: dict, *, extractor: str, transcript_source: str | None, + run_meta: dict | None, trades: list[dict], passed: list[dict]) -> None: + """rec: the extraction record; trades/passed: rec items enriched with symbol/resolve/our_view + (and entry_time_tape for trades).""" + vid = rec["video_id"] + sd = _date(rec.get("session_date")) or _date(rec.get("upload_date")) + day = rec.get("day") or {} + with connect() as conn: + with conn.cursor() as cur: + cur.execute( + f"""INSERT INTO {S}.recap_videos (video_id, upload_date, session_date, title, duration_s, + video_kind, transcript_source, extractor, status, last_error, run_meta, record, processed_at) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,'done',NULL,%s,%s,%s) + ON CONFLICT (video_id) DO UPDATE SET upload_date=EXCLUDED.upload_date, + session_date=EXCLUDED.session_date, title=EXCLUDED.title, duration_s=EXCLUDED.duration_s, + video_kind=EXCLUDED.video_kind, transcript_source=EXCLUDED.transcript_source, + extractor=EXCLUDED.extractor, status='done', last_error=NULL, attempts=0, run_meta=EXCLUDED.run_meta, + record=EXCLUDED.record, processed_at=EXCLUDED.processed_at, updated_at=now()""", + (vid, _date(rec.get("upload_date")), sd, meta.get("title"), + int(meta["duration"]) if str(meta.get("duration") or "").isdigit() else None, + rec.get("video_kind"), transcript_source, extractor, _j(run_meta), _j(rec), + datetime.now(timezone.utc))) + for t in ("recap_day_summary", "recap_trades", "recap_passed_on", "recap_rules"): + cur.execute(f"DELETE FROM {S}.{t} WHERE video_id=%s", (vid,)) + cur.execute( + f"""INSERT INTO {S}.recap_day_summary (video_id, session_date, net_pnl, green, market_temperature, + market_quote, max_loss_hit, daily_goal_comment, sizing_comment) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + (vid, sd, day.get("net_pnl"), day.get("green"), day.get("market_temperature"), + day.get("market_quote"), day.get("max_loss_hit"), day.get("daily_goal_comment"), + day.get("sizing_comment"))) + for i, t in enumerate(trades): + cur.execute( + f"""INSERT INTO {S}.recap_trades (video_id, seq, session_date, ticker_heard, ticker_guess, symbol, + resolve, setup, setup_quote, entry_price, exit_price, stated_high, stated_low, shares, pnl, + outcome, add_on, entry_time_hint, entry_time_tape, stock_context, why_entered, tape_reads, + why_exited, mistake_or_lesson, our_view) + VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + (vid, i, sd, t.get("ticker_heard"), t.get("ticker_guess"), t.get("symbol"), t.get("resolve"), + t.get("setup"), t.get("setup_quote"), t.get("entry_price"), t.get("exit_price"), + t.get("stated_high"), t.get("stated_low"), t.get("shares"), t.get("pnl"), t.get("outcome"), + bool(t.get("add_on")), t.get("entry_time_hint"), t.get("entry_time_tape"), + _j(t.get("stock_context")), _j(t.get("why_entered")), _j(t.get("tape_reads")), + _j(t.get("why_exited")), _j(t.get("mistake_or_lesson")), _j(t.get("our_view")))) + for i, p in enumerate(passed): + cur.execute( + f"""INSERT INTO {S}.recap_passed_on (video_id, seq, session_date, ticker_heard, ticker_guess, + symbol, resolve, reason, quote, our_view) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)""", + (vid, i, sd, p.get("ticker_heard"), p.get("ticker_guess"), p.get("symbol"), p.get("resolve"), + p.get("reason"), p.get("quote"), _j(p.get("our_view")))) + for i, r in enumerate(rec.get("rules_stated") or []): + if r.get("quote"): + cur.execute(f"INSERT INTO {S}.recap_rules (video_id, seq, rule, quote) VALUES (%s,%s,%s,%s)", + (vid, i, r.get("rule"), r["quote"])) + conn.commit() + + +def load_rows() -> tuple[list[dict], list[dict], list[dict]]: + """(videos, trades, passed) for the report -- every processed video.""" + with connect() as conn, conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) as cur: + cur.execute(f"""SELECT v.video_id, v.upload_date, v.session_date, v.title, v.video_kind, v.extractor, + d.net_pnl, d.green, d.market_temperature, d.market_quote + FROM {S}.recap_videos v LEFT JOIN {S}.recap_day_summary d USING (video_id) + WHERE v.status='done' ORDER BY v.upload_date, v.video_id""") + videos = [dict(r) for r in cur.fetchall()] + cur.execute(f"SELECT * FROM {S}.recap_trades ORDER BY session_date, video_id, seq") + trades = [dict(r) for r in cur.fetchall()] + cur.execute(f"SELECT * FROM {S}.recap_passed_on ORDER BY session_date, video_id, seq") + passed = [dict(r) for r in cur.fetchall()] + return videos, trades, passed diff --git a/youtube_claims/recap/tape.py b/youtube_claims/recap/tape.py new file mode 100644 index 0000000..10a9320 --- /dev/null +++ b/youtube_claims/recap/tape.py @@ -0,0 +1,209 @@ +"""Ticker verification against the tape + tape stats (ported from the pilot's join.py/ourside.py). + +A transcript ticker is only trusted when that symbol's SIP minute tape that day can produce the +story he told: his entry price traded, then (if he named one) a high near his stated high, then +(if named) a low near his stated low. Otherwise we search the day's scanner names for the one +unique symbol whose tape fits. The scoring/resolution is pure -- bars come from an injected +`bars_for(symbol)` callable, so it is unit-tested with fake bars.""" + +from __future__ import annotations + +import datetime as dt +import json +import re +from pathlib import Path +from typing import Callable, Sequence + +# a bar is (ts_et naive datetime, open, high, low, close, volume) +Bar = tuple +_SYM = re.compile(r"[A-Z]{1,5}") + + +def path_score(bars: Sequence[Bar], entry: float | None, sh: float | None, sl: float | None) -> float | None: + """None if the tape can't produce his story; else a closeness score (lower = better). + Story: entry trades (+-2% of the day's range), then a high within 2.5% of stated_high, + then a low within 3% of stated_low.""" + if not bars: + return None + lo = min(b[3] for b in bars) + hi = max(b[2] for b in bars) + if entry is not None and not (lo * 0.98 <= entry <= hi * 1.02): + return None + if sh is None: + return 0.5 if entry is not None else None + t_entry = next((i for i, b in enumerate(bars) + if entry is None or (b[3] <= entry * 1.01 and b[2] >= entry * 0.99)), None) + if t_entry is None: + return None + best = None + for i in range(t_entry, len(bars)): + err = abs(bars[i][2] - sh) / sh + if err > 0.025: + continue + e2 = err + if sl is not None: + later = [abs(b[3] - sl) / sl for b in bars[i:]] + if not later or min(later) > 0.03: + continue + e2 += min(later) + best = e2 if best is None else min(best, e2) + return best + + +def loose_fit(bars: Sequence[Bar], entry: float | None, sh: float | None, sl: float | None) -> bool: + """Order-free check: every level he named traded at SOME point that day (entry inside the + day's range, a bar high within 2.5% of stated_high, a bar low within 3% of stated_low). + Used only for the ticker he NAMED -- the name is a strong prior, and he often tells the + flush-then-squeeze out of order ("squeezed to 5 ... it had come from 2.35").""" + if not bars or (entry is None and sh is None): + return False + lo, hi = min(b[3] for b in bars), max(b[2] for b in bars) + if entry is not None and not (lo * 0.98 <= entry <= hi * 1.02): + return False + if sh is not None and not any(abs(b[2] - sh) / sh <= 0.025 for b in bars): + return False + if sl is not None and not any(abs(b[3] - sl) / sl <= 0.03 for b in bars): + return False + return True + + +def resolve(item: dict, bars_for: Callable[[str], Sequence[Bar]], scanner_syms: Sequence[str]) -> tuple[str | None, str]: + """(symbol, how). how: price_verified (his name fits the tape in order) | price_verified_loose + (his name has all his levels, out of order) | price_matched_scanner (a unique scanner name + fits in order) | ambiguous:A,B | no_price_match | unverified (no price given).""" + heard = (item.get("ticker_heard") or "").upper().replace(" ", "") + guess = (item.get("ticker_guess") or "").upper().replace(" ", "") + entry, sh, sl = item.get("entry_price"), item.get("stated_high"), item.get("stated_low") + cands = [c for c in dict.fromkeys([guess, heard]) if c and _SYM.fullmatch(c)] + if entry is None and sh is None: + return (cands[0] if cands else None), "unverified" + for c in cands: + if path_score(bars_for(c), entry, sh, sl) is not None: + return c, "price_verified" + for c in cands: + if loose_fit(bars_for(c), entry, sh, sl): + return c, "price_verified_loose" + scored = [] + for s in dict.fromkeys(scanner_syms): + if s in cands: + continue + sc = path_score(bars_for(s), entry, sh, sl) + if sc is not None: + scored.append((sc, s)) + scored.sort() + if len(scored) == 1 or (len(scored) > 1 and scored[0][0] + 0.01 < scored[1][0]): + return scored[0][1], "price_matched_scanner" + if scored: + return None, "ambiguous:" + ",".join(s for _, s in scored[:4]) + return (cands[0] if cands else None), "no_price_match" + + +def entry_time(bars: Sequence[Bar], entry: float | None, sh: float | None) -> dt.datetime | None: + """First minute the tape traded his entry price, before his stated high (if any).""" + if entry is None or not bars: + return None + stop = len(bars) + if sh: + hi_i = next((i for i, b in enumerate(bars) if abs(b[2] - sh) / sh <= 0.025), None) + if hi_i is not None: + stop = hi_i + 1 + for b in bars[:stop]: + if b[3] <= entry * 1.005 and b[2] >= entry * 0.995: + return b[0] + return None + + +def entry_bucket(t: dt.datetime | None) -> str | None: + if t is None: + return None + tt = t.time() + return ("pre-7:00" if tt < dt.time(7) else "7:00-9:30" if tt < dt.time(9, 30) + else "9:30-9:45" if tt < dt.time(9, 45) else "9:45+") + + +def tape_stats(bars: Sequence[Bar], prev_close: float | None) -> dict: + """Premarket/RTH summary of one symbol-day (times are ET).""" + out: dict = {"prev_close": prev_close} + if not bars: + return out + pre = [b for b in bars if b[0].time() < dt.time(9, 30)] + rth = [b for b in bars if dt.time(9, 30) <= b[0].time() < dt.time(16, 0)] + if pre: + out.update(pm_high=max(b[2] for b in pre), pm_low=min(b[3] for b in pre), + pm_volume=sum(b[5] for b in pre)) + if rth: + o = rth[0][1] + hod = max(rth, key=lambda b: b[2]) + out.update(open=o, hod=hod[2], hod_time=hod[0].strftime("%H:%M"), lod=min(b[3] for b in rth), + close=rth[-1][4], rth_volume=sum(b[5] for b in rth), + gap_pct=(o / prev_close - 1) if prev_close else None, + open_to_hod_pct=hod[2] / o - 1 if o else None) + top = max(bars, key=lambda b: b[2]) + out["day_high"] = top[2] + out["day_high_time"] = top[0].strftime("%H:%M") + first = next((b for b in bars if prev_close and b[2] >= prev_close * 1.3), None) + out["first_30pct_time"] = first[0].strftime("%H:%M") if first else None + return out + + +class BarSource: + """Complete SIP minute tape (04:00-20:00 ET) from Alpaca, cached on disk as JSON per + symbol-day (bars of a finished day never change). Falls back to our own minute_bars + when Alpaca has nothing -- ours miss the pre-discovery minutes of late-found names.""" + + def __init__(self, cache_dir: Path, paper_con=None): + self.cache_dir = Path(cache_dir) + self.paper_con = paper_con + self._mem: dict = {} + self._client = None + + def _alpaca(self): + if self._client is None: + from alpaca_paper.client import AlpacaPaperClient + from alpaca_paper.settings import AlpacaPaperSettings + self._client = AlpacaPaperClient(AlpacaPaperSettings.from_env()) + return self._client + + def bars(self, sym: str, d: str) -> list[Bar]: + key = (sym, str(d)) + if key in self._mem: + return self._mem[key] + f = self.cache_dir / str(d) / f"{sym}.json" + if f.exists(): + rows = json.loads(f.read_text()) + out = [(dt.datetime.fromisoformat(r[0]), *r[1:]) for r in rows] + self._mem[key] = out + return out + out = self._fetch(sym, str(d)) + if out or dt.date.fromisoformat(str(d)) < dt.date.today(): + f.parent.mkdir(parents=True, exist_ok=True) + f.write_text(json.dumps([(b[0].isoformat(), *b[1:]) for b in out])) + self._mem[key] = out + return out + + def _fetch(self, sym: str, d: str) -> list[Bar]: + from zoneinfo import ZoneInfo + et = ZoneInfo("America/New_York") + day = dt.date.fromisoformat(d) + start = dt.datetime.combine(day, dt.time(4), et) + end = dt.datetime.combine(day, dt.time(20), et) + raw = [] + try: + raw = self._alpaca().get_minute_bars( + [sym], start_iso=start.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), + end_iso=end.astimezone(dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"), feed="sip").get(sym) or [] + except Exception: # noqa: BLE001 -- unknown/delisted symbol, network: fall back below + raw = [] + out = [] + for b in raw: + ts = dt.datetime.fromisoformat(b["t"].replace("Z", "+00:00")).astimezone(et).replace(tzinfo=None) + out.append((ts, float(b["o"]), float(b["h"]), float(b["l"]), float(b["c"]), float(b["v"]))) + if out or self.paper_con is None: + return out + with self.paper_con.cursor() as cur: # our own bars (UTC-naive) as a fallback + cur.execute("select timestamp, open, high, low, close, volume from minute_bars " + "where symbol=%s and session_date=%s order by timestamp", (sym, d)) + rows = cur.fetchall() + utc = dt.timezone.utc + return [(r[0].replace(tzinfo=utc).astimezone(et).replace(tzinfo=None), float(r[1]), float(r[2]), + float(r[3]), float(r[4]), float(r[5] or 0)) for r in rows] diff --git a/youtube_claims/transcription.py b/youtube_claims/transcription.py index 07e7847..491e5d2 100644 --- a/youtube_claims/transcription.py +++ b/youtube_claims/transcription.py @@ -40,10 +40,15 @@ def _transcribe_whisper(wav: str, watchlist: list[str]) -> list[dict]: def _captions_fallback(video_id: str) -> list[dict]: + """YouTube captions via youtube-transcript-api. The API shape is chosen by feature + detection, NOT by catching an exception: the old code retried the removed pre-1.0 + `get_transcript` after ANY fetch failure, so the real cause (IpBlocked / RequestBlocked + when YouTube blocks the server IP) was replaced by a misleading AttributeError. Errors now + propagate with their real type and message.""" from youtube_transcript_api import YouTubeTranscriptApi - try: # newer + older API shapes + if hasattr(YouTubeTranscriptApi, "fetch"): # >= 1.0: instance API rows = YouTubeTranscriptApi().fetch(video_id).to_raw_data() - except Exception: # noqa: BLE001 + else: # < 1.0: classmethod API rows = YouTubeTranscriptApi.get_transcript(video_id) return [{"start": round(x["start"], 2), "end": round(x["start"] + x.get("duration", 0), 2), "text": x["text"]} for x in rows] @@ -55,6 +60,7 @@ def transcribe(video_id: str, watchlist: list[str] | None = None) -> tuple[list[ watchlist = watchlist if watchlist is not None else config.watchlist() os.makedirs(config.TRANSCRIPT_DIR, exist_ok=True) segs, source, model = None, None, None + whisper_err = None try: with tempfile.TemporaryDirectory() as td: wav = _download_audio(video_id, td) @@ -62,9 +68,14 @@ def transcribe(video_id: str, watchlist: list[str] | None = None) -> tuple[list[ segs = _transcribe_whisper(wav, watchlist) source, model = "whisper", config.WHISPER_MODEL except Exception as exc: # noqa: BLE001 — fall through to captions - print(f"[transcribe] whisper path failed for {video_id}: {exc}") + whisper_err = f"{type(exc).__name__}: {exc}" + print(f"[transcribe] whisper path failed for {video_id}: {whisper_err}") if not segs: - segs = _captions_fallback(video_id) # may raise -> worker retry/backoff + try: + segs = _captions_fallback(video_id) + except Exception as exc: # noqa: BLE001 — report BOTH causes (worker retry/backoff) + raise RuntimeError(f"whisper path: {whisper_err or 'no audio file produced'}; " + f"captions: {type(exc).__name__}: {str(exc)[:500]}") from exc source = "captions" path = os.path.join(config.TRANSCRIPT_DIR, f"{video_id}.json") with open(path, "w") as f: