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: