finetune_data.py

"""
finetune_data.py — capture (instruction, input, output) pairs from live app use,
then export a clean JSONL ready for LoRA SFT (the 🎯 Well-Tuned badge).

Every time the Signals "AI summary" runs, app.py calls `record()` with the
English raw read (input) and the model's narrative (output). Pairs accumulate
on the /data bucket and survive restarts; the Model tab has an "Export dataset"
button that writes a timestamped JSONL and reports the count.

The dataset teaches a small model ONE focused skill — turn a Chan-theory raw
read into a crisp long-hold trading summary — which is exactly what the app's
Translator sub-agent does. A 1.7B model fine-tuned on this beats a generic 4B
at the task, and doubles as the Tiny Titan entry.
"""
from __future__ import annotations

import json
import os
import re
import threading
import datetime as dt

import paths

_PAIRS = os.path.join(paths.DATASET_DIR, "pairs.jsonl")
_lock = threading.Lock()

INSTRUCTION = ("You are an equity analyst. Based only on this factual read of a "
               "US stock's multi-timeframe Chan-theory verdict, write a short "
               "plain-English summary for a long-term holder: the situation "
               "today, whether to act or wait, and the key price levels. "
               "Max 90 words, no disclaimers.")


def _clean(text: str) -> str:
    # strip stray think tags and any "AI narrative ..." UI prefix
    text = re.sub(r"<think>.*?</think>", "", text, flags=re.S)
    text = re.sub(r"^🤖\s*\*\*AI narrative[^\n]*\*\*\s*", "", text)
    text = text.replace("🤖 **AI narrative (Translator sub-agent · Qwen3-1.7B):**", "")
    return text.strip()


def record(raw_read: str, narrative: str):
    """Append one training pair. Silently ignores junk / unloaded-model output."""
    narrative = _clean(narrative)
    if not raw_read or not narrative or len(narrative) < 40:
        return
    if narrative.startswith(("⏳", "(", "Run the analysis")):
        return
    row = {"instruction": INSTRUCTION, "input": raw_read.strip(), "output": narrative}
    try:
        with _lock, open(_PAIRS, "a", encoding="utf-8") as f:
            f.write(json.dumps(row, ensure_ascii=False) + "\n")
    except OSError:
        pass


def count() -> int:
    try:
        with open(_PAIRS, encoding="utf-8") as f:
            return sum(1 for _ in f)
    except OSError:
        return 0


def export() -> str:
    """De-duplicate and write a timestamped JSONL. Returns the file path so the
    UI can offer it as a direct download."""
    n = count()
    if n == 0:
        return ""
    seen, rows = set(), []
    try:
        with open(_PAIRS, encoding="utf-8") as f:
            for line in f:
                line = line.strip()
                if not line:
                    continue
                try:
                    r = json.loads(line)
                except ValueError:
                    continue
                key = (r.get("input", ""), r.get("output", ""))
                if key in seen:
                    continue
                seen.add(key)
                rows.append(r)
    except OSError:
        return ""
    stamp = dt.datetime.utcnow().strftime("%Y%m%d-%H%M%S")
    out = os.path.join(paths.DATASET_DIR, f"chan_sft_{stamp}.jsonl")
    try:
        with open(out, "w", encoding="utf-8") as f:
            for r in rows:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
    except OSError:
        return ""
    return out


def status_line() -> str:
    n = count()
    if n == 0:
        return "_No fine-tuning pairs captured yet — run a few Signals AI summaries._"
    return f"📚 **{n}** training pair(s) captured on /data (target: 200-500 for a good LoRA)."