Capstone C4: una plataforma de investigación con LLMs y datos alternativos.

import hashlib
import json
import re
import time

import numpy as np
import pandas as pd
from scipy.stats import norm

def huella(ruta):
    """SHA-256 abreviado de un fichero de datos (va al informe)."""
    with open(ruta, "rb") as f:
        return hashlib.sha256(f.read()).hexdigest()[:12]

FICHEROS = ["acciones6_precios.parquet", "noticias6_corpus.parquet",
            "acciones6_tickers.csv", "texto6_lexico.csv",
            "rag6_documentos.parquet", "rag6_pasajes.parquet",
            "rag6_preguntas.csv"]
HUELLAS = {f: huella(f) for f in FICHEROS}

precios = pd.read_parquet("acciones6_precios.parquet")
C = precios.pivot(index="fecha", columns="ticker", values="cierre")
R = C.pct_change().to_numpy()
S = C.index
noticias = pd.read_parquet("noticias6_corpus.parquet")
historia = pd.read_csv("acciones6_tickers.csv")
empresas = pd.read_csv("acciones6_empresas.csv")
actual = empresas.set_index("id_empresa")["ticker"]
noticias["tk"] = noticias["ticker"].map(dict(zip(
    historia["ticker"], historia["id_empresa"].map(actual))))

# Cierre: 16:00 de Nueva York; 13:00 en las medias sesiones
# (exchange_calendars 4.13.2, XNYS, como en la lección 6.4.6)
MEDIAS = pd.to_datetime([
    "2019-07-03", "2019-11-29", "2019-12-24", "2020-11-27",
    "2020-12-24", "2021-11-26", "2022-11-25", "2023-07-03",
    "2023-11-24", "2024-07-03", "2024-11-29", "2024-12-24",
    "2025-07-03", "2025-11-28", "2025-12-24"])
hora = np.where(S.isin(MEDIAS), 13, 16)
cierres = (S + pd.to_timedelta(hora, unit="h")).tz_localize(
    "America/New_York").tz_convert("UTC")
CORTE = pd.Timedelta(minutes=30)     # tiempo para decidir y enviar
ts = noticias["ts"].to_numpy()
noticias["k"] = np.searchsorted((cierres - CORTE).to_numpy(), ts,
                                side="right")
k_reac = np.searchsorted(cierres.to_numpy(), ts, side="right")

verdad = pd.read_csv("noticias6_verdad.csv")
v = verdad.set_index("id_noticia").loc[noticias["id_noticia"]]
con = v["sesion_reaccion"].notna().to_numpy() & (k_reac < len(S))
calc = S[np.minimum(k_reac, len(S) - 1)].strftime("%Y-%m-%d")
igual = calc[con] == v["sesion_reaccion"].to_numpy()[con]
print(f"sesión de reacción = verdad: {igual.mean():.1%} "
      f"de {con.sum()}")
print(f"pasan a la sesión siguiente por el corte de 30 min: "
      f"{(noticias['k'].to_numpy() != k_reac).mean():.1%}")
print(HUELLAS["noticias6_corpus.parquet"],
      noticias["tk"].isna().sum(), "sin empresa")

from transformers import pipeline

lex = pd.read_csv("texto6_lexico.csv").query("lista == 'financiera'")
POS = set(lex.loc[lex["categoria"] == "positiva", "palabra"])
NEG = set(lex.loc[lex["categoria"] == "negativa", "palabra"])

def tono_dicc(texto):
    w = re.findall(r"[a-záéíóúüñ]+", texto.lower())
    p, n = sum(x in POS for x in w), sum(x in NEG for x in w)
    return (p - n) / (p + n + 1)

class Cache:
    """Puntuaciones por (modelo, texto): se calculan una vez."""
    def __init__(self, ruta):
        self.ruta, self.d = ruta, {}
    def clave(self, modelo, texto):
        return hashlib.sha256(f"{modelo}\0{texto}".encode()).hexdigest()
    def puntuar(self, modelo, textos, f):
        claves = [self.clave(modelo, t) for t in textos]
        nuevas = [i for i, c in enumerate(claves) if c not in self.d]
        if nuevas:
            for i, s in zip(nuevas, f([textos[i] for i in nuevas])):
                self.d[claves[i]] = float(s)
        pd.Series(self.d).rename("valor").to_frame().to_parquet(
            self.ruta)
        return np.array([self.d[c] for c in claves]), len(nuevas)

finbert = pipeline("text-classification", model="ProsusAI/finbert",
                   device="cpu")
def f_finbert(textos):
    out = finbert(textos, batch_size=32, top_k=None, truncation=True)
    p = [{d["label"]: d["score"] for d in s} for s in out]
    return [x["positive"] - x["negative"] for x in p]

cache = Cache("puntuaciones.parquet")
texto = (noticias["titular"] + " " + noticias["cuerpo"]).tolist()
noticias["dicc"], n1 = cache.puntuar(
    "dicc-v1", texto, lambda ts: [tono_dicc(t) for t in ts])
en = noticias.index[noticias["idioma"] == "en"]
muestra = np.random.default_rng(765).choice(en, 200, replace=False)
fb, n2 = cache.puntuar("finbert@0bc2ac1",
                       [texto[i] for i in muestra], f_finbert)
_, n3 = cache.puntuar("finbert@0bc2ac1",
                      [texto[i] for i in muestra], f_finbert)
sv = v["sentimiento_verdadero"].to_numpy()[muestra]
dm = noticias["dicc"].to_numpy()[muestra]
print(f"calculadas: dicc {n1}, finbert {n2}; repetición: {n3}")
print(f"correlación con la verdad (200 en inglés): finbert "
      f"{np.corrcoef(fb, sv)[0, 1]:.3f}, diccionario "
      f"{np.corrcoef(dm, sv)[0, 1]:.3f}")

def panel(valores):
    """Suma por sesión (con el corte) y acción, sin mirar al futuro."""
    ok = noticias["k"].to_numpy() < len(S)
    d = pd.DataFrame({"k": noticias["k"][ok], "tk": noticias["tk"][ok],
                      "x": np.asarray(valores)[ok]})
    return (d.groupby(["k", "tk"])["x"].sum().unstack()
            .reindex(index=range(len(S)), columns=C.columns)
            .fillna(0.0).to_numpy())

def cartera(P, h=5, umbral=0.0, coste=5e-4):
    """Larga en las acciones con tono acumulado de h sesiones > umbral,
    corta en las < −umbral; 50 % por lado; rinde la sesión siguiente."""
    s = pd.DataFrame(P).rolling(h, min_periods=1).sum().to_numpy()
    lar, cor = s > umbral, s < -umbral
    nl = lar.sum(1, keepdims=True)
    nc = cor.sum(1, keepdims=True)
    w = np.where((nl > 0) & (nc > 0),
                 0.5 * lar / np.maximum(nl, 1)
                 - 0.5 * cor / np.maximum(nc, 1), 0.0)
    bruto = np.nansum(w[:-1] * R[1:], axis=1)
    giro = np.abs(np.diff(w, axis=0)).sum(1)
    r = pd.Series(bruto - coste * giro, index=S[1:])
    return r[r.index >= "2019-01-07"], giro.mean()

def sharpe(r):
    return float(r.mean() / r.std() * np.sqrt(252))

P_dicc = panel(noticias["dicc"])
P_orac = panel(v["sentimiento_verdadero"].to_numpy()
               * v["materialidad"].to_numpy())
filas = {}
for nombre, P in (("diccionario", P_dicc), ("oráculo", P_orac)):
    for c in (0.0, 5e-4, 10e-4):
        r, giro = cartera(P, 5, 0.0, c)
        filas[(nombre, f"{c * 1e4:.0f} pb")] = {
            "Sharpe": sharpe(r), "anual %": 252 * 100 * r.mean(),
            "giro diario": giro}
pd.DataFrame(filas).T.round(2)

import os

REG = "experimentos/registro_pruebas.jsonl"    # solo de añadir (7.6.1)
os.makedirs("experimentos", exist_ok=True)

def registrar(config, r, hipotesis):
    """Una línea por prueba; en el proyecto, también fecha y commit."""
    cfg = json.dumps(config, sort_keys=True)
    with open(REG, "a", encoding="utf-8") as f:
        f.write(json.dumps({
            "id": hashlib.sha256(cfg.encode()).hexdigest()[:10],
            "hipotesis": hipotesis, "config": config,
            "datos": HUELLAS["noticias6_corpus.parquet"],
            "resultado": {"sharpe": round(sharpe(r), 4), "n": len(r)}},
            ensure_ascii=False) + "\n")

REJILLA = [(h, u) for h in (1, 3, 5, 10) for u in (0.0, 0.3)]
series = {}
for h, u in REJILLA:
    series[(h, u)] = cartera(P_dicc, h, u)[0]
    registrar({"señal": "dicc-v1", "h": h, "umbral": u,
               "coste_pb": 5}, series[(h, u)],
              "el tono anticipa la deriva de h sesiones")
with open(REG, encoding="utf-8") as f:
    N = sum(1 for _ in f)

oos = []
for ano in range(2021, 2026):
    ent = {k: sharpe(x[x.index.year < ano]) for k, x in series.items()}
    elegida = max(ent, key=ent.get)
    x = series[elegida]
    x = x[x.index.year == ano]
    if ent[elegida] <= 0:              # regla fijada antes: sin
        x = x * 0.0                    # ventaja en entrenamiento,
    oos.append(x)                      # ese año no se opera
    fuera = (f"{sharpe(x):.2f}" if ent[elegida] > 0
             else "no se opera")
    print(ano, "elegida (h, umbral) =", elegida,
          f"| entrena {ent[elegida]:.2f} | fuera {fuera}")
oos = pd.concat(oos)

def max_esperado(n):
    g = np.euler_gamma
    return ((1 - g) * norm.ppf(1 - 1 / n)
            + g * norm.ppf(1 - 1 / (n * np.e)))

def psr(x, umbral_diario):
    sr = x.mean() / x.std()
    a, k = x.skew(), x.kurt() + 3
    den = np.sqrt(1 - a * sr + (k - 1) / 4 * sr ** 2)
    return float(norm.cdf((sr - umbral_diario) * np.sqrt(len(x) - 1)
                          / den))

sr_d = pd.Series({k: x.mean() / x.std() for k, x in series.items()})
mejor = max(series, key=lambda k: sharpe(series[k]))
sr0 = sr_d.std() * max_esperado(N)
print(f"N = {N} pruebas; mejor {mejor}: Sharpe "
      f"{sharpe(series[mejor]):.2f}, umbral "
      f"{sr0 * np.sqrt(252):.2f}, DSR {psr(series[mejor], sr0):.3f}")
print(f"walk-forward 2021-2025: Sharpe {sharpe(oos):.2f}, "
      f"{252 * 100 * oos.mean():.2f} % anual")

import faiss
import unicodedata
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer

docs = pd.read_parquet("rag6_documentos.parquet")
pas = pd.read_parquet("rag6_pasajes.parquet").query("tipo != 'titulo'")
pas = pas.reset_index(drop=True)
preg = pd.read_csv("rag6_preguntas.csv")
TIT = dict(zip(docs["id_doc"], docs["titulo"]))
indexado = [f"{TIT[d]} > {s}\n{t}" for d, s, t in
            zip(pas["id_doc"], pas["seccion"], pas["texto"])]

def palabras(s):
    s = unicodedata.normalize("NFKD", s.lower())
    return re.findall(r"\w+", "".join(
        c for c in s if not unicodedata.combining(c)))

bm = BM25Okapi([palabras(t) for t in indexado])
emb = SentenceTransformer("sentence-transformers/all-MiniLM-L6-v2",
                          device="cpu", local_files_only=True)
X = emb.encode(indexado, normalize_embeddings=True, batch_size=64,
               show_progress_bar=False).astype("float32")
indice = faiss.IndexFlatIP(X.shape[1])
indice.add(X)

def ranking(pregunta, modo):
    if modo == "bm25":
        return np.argsort(-bm.get_scores(palabras(pregunta)),
                          kind="stable")
    q = emb.encode([pregunta], normalize_embeddings=True,
                   show_progress_bar=False).astype("float32")
    _, I = indice.search(q, len(pas))
    if modo == "minilm":
        return I[0]
    rrf = np.zeros(len(pas))           # fusión de rankings (6.3.1)
    for orden in (ranking(pregunta, "bm25"), I[0]):
        rrf[orden] += 1 / (60 + np.arange(1, len(pas) + 1))
    return np.argsort(-rrf, kind="stable")

ids = pas["id_pasaje"].to_numpy()
resp = preg[preg["respondible"]]
for modo in ("bm25", "minilm", "fusión"):
    rec = [np.mean([o in set(ids[ranking(q.pregunta, modo)[:3]])
                    for o in q.pasajes_oro.split(";")])
           for q in resp.itertuples()]
    print(f"{modo:<7} recall@3 = {np.mean(rec):.3f}")

import operator
from typing import Annotated, TypedDict

import torch
from langgraph.graph import END, START, StateGraph
from transformers import AutoModelForCausalLM, AutoTokenizer

torch.manual_seed(0)
torch.use_deterministic_algorithms(True)
QWEN = "Qwen/Qwen2.5-0.5B-Instruct"
tok = AutoTokenizer.from_pretrained(QWEN)
llm = AutoModelForCausalLM.from_pretrained(QWEN, dtype=torch.float32)
SISTEMA = ("Respondes preguntas sobre informes usando SOLO los "
           "fragmentos numerados. Una frase, con la cifra exacta y la "
           "cita [n]. Si los fragmentos no lo dicen, responde "
           "exactamente: No consta.")

NUM = re.compile(r"\d[\d.,]*\d|\d")
CITA = re.compile(r"\[(\d+)\]")
def cifras(s):
    out = set()
    for t in NUM.findall(CITA.sub(" ", s)):
        if "," in t:
            t = t.replace(".", "").replace(",", ".")
        elif re.fullmatch(r"\d{1,3}(\.\d{3})+", t):
            t = t.replace(".", "")
        out.add(round(float(t.rstrip(".")), 4))
    return out

class Estado(TypedDict, total=False):
    pregunta: str
    top: list
    textos: list
    borrador: str
    veredicto: str
    cita: str
    traza: Annotated[list, operator.add]

def recuperar(s):
    top = ranking(s["pregunta"], "bm25")[:3]
    return {"top": [ids[j] for j in top],
            "textos": [pas["texto"][j] for j in top],
            "traza": ["recuperar"]}

def redactar(s):
    ctx = "\n\n".join(f"[{i + 1}] {t[:600]}"
                      for i, t in enumerate(s["textos"]))
    m = [{"role": "system", "content": SISTEMA},
         {"role": "user", "content": f"{ctx}\n\nPregunta: "
                                     f"{s['pregunta']}"}]
    x = tok.apply_chat_template(m, add_generation_prompt=True,
                                return_tensors="pt", return_dict=True)
    with torch.inference_mode():
        y = llm.generate(**x, do_sample=False, max_new_tokens=48,
                         pad_token_id=tok.eos_token_id)
    txt = tok.decode(y[0, x["input_ids"].shape[1]:],
                     skip_special_tokens=True)
    return {"borrador": txt.strip().replace("`", "'"),
            "traza": ["redactar"]}

def validar(s):
    """La cita la pone el código: el fragmento que contiene TODAS las
    cifras del borrador. Sin cifras o sin fragmento, no se publica."""
    b = s["borrador"]
    if b.startswith("No consta"):
        return {"veredicto": "abstención", "traza": ["validar"]}
    dichas = cifras(b) - cifras(s["pregunta"])
    if not dichas:
        return {"veredicto": "rechazo: sin cifra",
                "traza": ["validar"]}
    for pid, t in zip(s["top"], s["textos"]):
        if dichas <= cifras(t):
            return {"veredicto": "publicable", "cita": pid,
                    "traza": ["validar"]}
    return {"veredicto": "rechazo: cifra sin fuente",
            "traza": ["validar"]}

def publicar(s):          # aquí iría la aprobación humana (6.4.2)
    return {"traza": [f"publicar {s['cita']}"]}

g = StateGraph(Estado)
for nombre, f in (("recuperar", recuperar), ("redactar", redactar),
                  ("validar", validar), ("publicar", publicar)):
    g.add_node(nombre, f)
g.add_edge(START, "recuperar")
g.add_edge("recuperar", "redactar")
g.add_edge("redactar", "validar")
g.add_conditional_edges(
    "validar", lambda s: s["veredicto"] == "publicable",
    {True: "publicar", False: END})
g.add_edge("publicar", END)
analista = g.compile()

N_TIPO = {"texto": 2, "tabla": 1, "nota_pie": 1, "calculo": 1,
          "no_respondible": 2}
muestra_q = pd.concat([preg[preg["tipo_pregunta"] == t]
                       .sample(n, random_state=5)
                       for t, n in N_TIPO.items()])
filas, trazas = [], {}
for q in muestra_q.itertuples():
    fin = analista.invoke({"pregunta": q.pregunta})
    trazas[q.id_pregunta] = fin["traza"]
    oro, b = cifras(q.respuesta_oro), fin["borrador"]
    bien = (b.startswith("No consta") if not q.respondible
            else bool(oro) and oro <= cifras(b))
    filas.append({"id": q.id_pregunta, "tipo": q.tipo_pregunta,
                  "veredicto": fin["veredicto"], "correcta": bien,
                  "cita": fin.get("cita", "")})
res_q = pd.DataFrame(filas).set_index("id")
print(res_q.to_string())
print(trazas["QA105"])

import anthropic

cliente = anthropic.Anthropic()          # ANTHROPIC_API_KEY del entorno

def redactar_remoto(s):
    ctx = "\n\n".join(f"[{i + 1}] {t}"
                      for i, t in enumerate(s["textos"]))
    r = cliente.messages.create(
        model="claude-opus-5-5", max_tokens=200, system=SISTEMA,
        messages=[{"role": "user", "content":
                   f"{ctx}\n\nPregunta: {s['pregunta']}"}])
    texto = r.content[0].text.strip()
    # se guarda con su clave (modelo, prompt, fragmentos): el informe
    # se reconstruye sin volver a llamar a la API (7.3.4)
    return {"borrador": texto, "traza": ["redactar_remoto"]}

publicadas = res_q[res_q["veredicto"] == "publicable"]
informe = {
    "proyecto": "C4", "datos": HUELLAS,
    "señal": {"elegida": {"h": mejor[0], "umbral": mejor[1]},
              "n_pruebas": N, "coste_pb": 5,
              "sharpe_completo": round(sharpe(series[mejor]), 2),
              "dsr": round(psr(series[mejor], sr0), 3),
              "sharpe_walk_forward": round(sharpe(oos), 2)},
    "analista": {
        "preguntas": len(res_q), "publicadas": len(publicadas),
        "publicadas_correctas": int(publicadas["correcta"].sum()),
        "abstenciones": int((res_q["veredicto"]
                             == "abstención").sum())},
}
informe["decision"] = ("seguir" if informe["señal"]["dsr"] >= 0.95
                       and informe["señal"]["sharpe_walk_forward"] > 0
                       else "parar o rediseñar")
with open("informe_c4.json", "w", encoding="utf-8") as f:
    json.dump(informe, f, ensure_ascii=False, indent=1)
for parte in ("señal", "analista", "decision"):
    print(parte, json.dumps(informe[parte], ensure_ascii=False))