Sync from GitHub (tests passed)

app/ai_engine.py

@@ -30,7 +30,16 @@ import xgboost as xgb
 from sklearn.metrics import mean_absolute_error, mean_squared_error
 
 from app.db import SessionLocal, init_db
-from app.models import NewsArticle, NewsSentiment, DailySentiment, PriceBar
+from app.models import (
+    NewsArticle,
+    NewsSentiment,
+    DailySentiment,
+    PriceBar,
+    NewsProcessed,
+    NewsRaw,
+    NewsSentimentV2,
+    DailySentimentV2,
+)
 from app.settings import get_settings
 from app.features import build_feature_matrix, get_feature_descriptions
 from app.lock import pipeline_lock
@@ -52,6 +61,116 @@ HYBRID_FALLBACK_PARSE_MODEL_NAME = "hybrid_fallback_parse"
 LLM_LABELS = {"BULLISH", "BEARISH", "NEUTRAL"}
 LLM_SCORING_MAX_TOKENS_PRIMARY = 2000
 LLM_SCORING_MAX_TOKENS_RETRY = 6000
+LLM_V2_LABEL_THRESHOLD = 0.15
+LLM_V2_EVENT_TYPES = {
+    "supply_disruption",
+    "supply_expansion",
+    "demand_increase",
+    "demand_decrease",
+    "inventory_draw",
+    "inventory_build",
+    "policy_support",
+    "policy_drag",
+    "macro_usd_up",
+    "macro_usd_down",
+    "cost_push",
+    "mixed_unclear",
+    "non_copper",
+}
+LLM_V2_EVENT_SIGN = {
+    "supply_disruption": 1,
+    "inventory_draw": 1,
+    "demand_increase": 1,
+    "policy_support": 1,
+    "macro_usd_down": 1,
+    "cost_push": 1,
+    "supply_expansion": -1,
+    "inventory_build": -1,
+    "demand_decrease": -1,
+    "policy_drag": -1,
+    "macro_usd_up": -1,
+    "mixed_unclear": 0,
+    "non_copper": 0,
+}
+LLM_V2_EVENT_STRENGTH = {
+    "supply_disruption": 1.0,
+    "inventory_draw": 0.9,
+    "demand_increase": 0.95,
+    "policy_support": 0.8,
+    "macro_usd_down": 0.7,
+    "cost_push": 0.75,
+    "supply_expansion": 1.0,
+    "inventory_build": 0.9,
+    "demand_decrease": 0.95,
+    "policy_drag": 0.8,
+    "macro_usd_up": 0.7,
+    "mixed_unclear": 0.25,
+    "non_copper": 0.0,
+}
+LLM_V2_SYSTEM_PROMPT = """You are a Senior Copper Futures Analyst focused on COMEX HG=F front-month contract.
+Your job is to estimate 1-5 trading day copper price impact from each article.
+
+Core principle:
+Classify by expected HG=F price reaction, NOT by whether the news is "good" or "bad" for the economy/company.
+
+Output requirements:
+Return ONLY a JSON array. One object per input id.
+Each object must contain exactly:
+- id (integer)
+- label ("BULLISH" | "BEARISH" | "NEUTRAL")
+- impact_score (number, -1.00 to 1.00, two decimals)
+- confidence (number, 0.00 to 1.00, two decimals)
+- relevance (number, 0.00 to 1.00, two decimals)
+- event_type (one of: supply_disruption, supply_expansion, demand_increase, demand_decrease, inventory_draw, inventory_build, policy_support, policy_drag, macro_usd_up, macro_usd_down, cost_push, mixed_unclear, non_copper)
+- reasoning (single line, <= 160 chars)
+
+Copper-specific reasoning rules:
+1) Supply tightening is typically BULLISH for copper price.
+2) Supply expansion is typically BEARISH.
+3) Demand increase is typically BULLISH.
+4) Demand decrease is typically BEARISH.
+5) USD stronger is usually BEARISH for dollar-denominated copper; USD weaker is usually BULLISH.
+6) If article is not materially related to copper supply/demand/pricing, use non_copper + NEUTRAL with low relevance/confidence.
+7) Use NEUTRAL only when net effect is truly mixed/unclear within 1-5 day horizon.
+
+Label mapping:
+- impact_score >= 0.15 => BULLISH
+- impact_score <= -0.15 => BEARISH
+- otherwise => NEUTRAL
+"""
+LLM_SCORING_RESPONSE_FORMAT_V2 = {
+    "type": "json_schema",
+    "json_schema": {
+        "name": "news_sentiment_scores_v2",
+        "strict": True,
+        "schema": {
+            "type": "array",
+            "items": {
+                "type": "object",
+                "properties": {
+                    "id": {"type": "integer"},
+                    "label": {"type": "string", "enum": ["BULLISH", "BEARISH", "NEUTRAL"]},
+                    "impact_score": {"type": "number", "minimum": -1, "maximum": 1},
+                    "confidence": {"type": "number", "minimum": 0, "maximum": 1},
+                    "relevance": {"type": "number", "minimum": 0, "maximum": 1},
+                    "event_type": {"type": "string", "enum": sorted(LLM_V2_EVENT_TYPES)},
+                    "reasoning": {"type": "string"},
+                },
+                "required": [
+                    "id",
+                    "label",
+                    "impact_score",
+                    "confidence",
+                    "relevance",
+                    "event_type",
+                    "reasoning",
+                ],
+                "additionalProperties": False,
+            },
+        },
+    },
+}
+SCORING_V2_VERSION = "commodity_v2"
 
 
 # =============================================================================
@@ -692,6 +811,837 @@ def score_batch_with_finbert(articles: list) -> list[dict]:
     return results
 
 
+def _clip(value: float, lower: float, upper: float) -> float:
+    """Clamp numeric value."""
+    return max(lower, min(upper, float(value)))
+
+
+def _label_from_impact_score(impact_score: float) -> str:
+    """Map impact score to discrete label."""
+    if impact_score >= LLM_V2_LABEL_THRESHOLD:
+        return "BULLISH"
+    if impact_score <= -LLM_V2_LABEL_THRESHOLD:
+        return "BEARISH"
+    return "NEUTRAL"
+
+
+def _sign(value: float, eps: float = 1e-9) -> int:
+    """Return numeric sign with epsilon deadzone."""
+    if value > eps:
+        return 1
+    if value < -eps:
+        return -1
+    return 0
+
+
+def _normalize_event_type(value: Any) -> str:
+    """Normalize event type to allowed vocabulary."""
+    normalized = str(value or "").strip().lower()
+    if normalized in LLM_V2_EVENT_TYPES:
+        return normalized
+    return "mixed_unclear"
+
+
+def _infer_event_type_from_text(text: str) -> str:
+    """Heuristic event inference used only for deterministic fallback."""
+    lower = (text or "").lower()
+    if not lower or "copper" not in lower:
+        return "non_copper"
+
+    supply_disruption_keywords = [
+        "outage",
+        "strike",
+        "disruption",
+        "shutdown",
+        "halt",
+        "sanction",
+        "inventory draw",
+        "stocks fell",
+        "warehouse draw",
+    ]
+    supply_expansion_keywords = [
+        "ramp-up",
+        "ramp up",
+        "increase output",
+        "production increase",
+        "new mine",
+        "inventory build",
+        "stocks rose",
+    ]
+    demand_increase_keywords = [
+        "stimulus",
+        "grid investment",
+        "ev demand",
+        "demand rise",
+        "stockpile purchase",
+        "import growth",
+    ]
+    demand_decrease_keywords = [
+        "slowdown",
+        "weak demand",
+        "demand decline",
+        "construction slump",
+        "pmi contraction",
+        "import decline",
+    ]
+    if any(token in lower for token in supply_disruption_keywords):
+        return "supply_disruption"
+    if any(token in lower for token in supply_expansion_keywords):
+        return "supply_expansion"
+    if any(token in lower for token in demand_increase_keywords):
+        return "demand_increase"
+    if any(token in lower for token in demand_decrease_keywords):
+        return "demand_decrease"
+    if "dollar strengthens" in lower or "usd stronger" in lower:
+        return "macro_usd_up"
+    if "dollar weakens" in lower or "usd weaker" in lower:
+        return "macro_usd_down"
+    return "mixed_unclear"
+
+
+def _build_llm_v2_user_prompt(articles: list[dict], horizon_days: int) -> str:
+    """Build compact JSON prompt for batch scoring."""
+    normalized_articles = [
+        {
+            "id": int(article["id"]),
+            "title": str(article.get("title") or "")[:500],
+            "description": str(article.get("description") or "")[:800],
+        }
+        for article in articles
+    ]
+    return (
+        f"Classify each article for {horizon_days}-day HG=F copper futures impact.\n"
+        "Return one object per id.\n\n"
+        f"Input articles JSON:\n{json.dumps(normalized_articles, ensure_ascii=True)}"
+    )
+
+
+def _parse_llm_v2_items(
+    *,
+    raw_results: Any,
+    expected_ids: list[int],
+    model_name: str,
+) -> tuple[dict[int, dict], list[int]]:
+    """
+    Parse/validate V2 LLM outputs.
+
+    Returns:
+        (valid_results_by_id, failed_ids)
+    """
+    if not isinstance(raw_results, list):
+        raise ValueError(f"Structured result must be a list, got {type(raw_results).__name__}")
+
+    expected = set(expected_ids)
+    valid: dict[int, dict] = {}
+    failed_ids: set[int] = set()
+
+    for item in raw_results:
+        if not isinstance(item, dict):
+            continue
+        if "id" not in item:
+            continue
+
+        try:
+            article_id = int(item["id"])
+        except (TypeError, ValueError):
+            continue
+
+        if article_id not in expected:
+            continue
+        if article_id in valid:
+            failed_ids.add(article_id)
+            continue
+
+        raw_label = item.get("label", item.get("classification"))
+        raw_impact = item.get("impact_score", item.get("score"))
+        raw_confidence = item.get("confidence")
+        raw_relevance = item.get("relevance", item.get("relevance_score"))
+        raw_event_type = item.get("event_type")
+        raw_reasoning = item.get("reasoning", "")
+
+        try:
+            if raw_impact is None:
+                raise ValueError("missing impact_score")
+            if raw_confidence is None:
+                raise ValueError("missing confidence")
+            if raw_relevance is None:
+                raise ValueError("missing relevance")
+
+            impact_score = _clip(float(raw_impact), -1.0, 1.0)
+            confidence = _clip(float(raw_confidence), 0.0, 1.0)
+            relevance = _clip(float(raw_relevance), 0.0, 1.0)
+        except (TypeError, ValueError):
+            failed_ids.add(article_id)
+            continue
+
+        event_type = _normalize_event_type(raw_event_type)
+        label_from_impact = _label_from_impact_score(impact_score)
+        if raw_label is None:
+            label = label_from_impact
+        else:
+            label = str(raw_label).upper().strip()
+            if label not in LLM_LABELS:
+                label = label_from_impact
+            if label != label_from_impact:
+                # Keep deterministic consistency between score and class.
+                label = label_from_impact
+
+        reasoning = _sanitize_reasoning_text(raw_reasoning)[:160]
+
+        valid[article_id] = {
+            "id": article_id,
+            "label": label,
+            "impact_score": impact_score,
+            "confidence": confidence,
+            "relevance": relevance,
+            "event_type": event_type,
+            "reasoning": reasoning,
+            "llm_model": model_name,
+        }
+
+    # Mark missing ids as failed.
+    for article_id in expected_ids:
+        if article_id not in valid:
+            failed_ids.add(article_id)
+
+    return valid, sorted(failed_ids)
+
+
+async def _repair_json_response_v2(
+    *,
+    settings: Any,
+    model_name: str,
+    malformed_content: str,
+    expected_ids: list[int],
+) -> str:
+    """Repair malformed JSON into V2 contract with no semantic rewrite."""
+    repair_prompt = (
+        "Convert the malformed output into valid JSON array.\n"
+        f"Expected ids: {expected_ids}\n"
+        "Keep keys: id,label,impact_score,confidence,relevance,event_type,reasoning.\n"
+        "Do not add explanations.\n\n"
+        f"MALFORMED:\n{malformed_content}"
+    )
+    repair_data = await create_chat_completion(
+        api_key=settings.openrouter_api_key,
+        model=model_name,
+        messages=[
+            {
+                "role": "system",
+                "content": "You repair JSON only. Return valid JSON array without markdown.",
+            },
+            {"role": "user", "content": repair_prompt},
+        ],
+        max_tokens=2600,
+        temperature=0.0,
+        timeout_seconds=60.0,
+        max_retries=settings.openrouter_max_retries,
+        rpm=settings.openrouter_rpm,
+        fallback_models=settings.openrouter_fallback_models_list,
+        referer="https://copper-mind.vercel.app",
+        title="CopperMind V2 JSON Repair",
+        extra_payload={"reasoning": {"exclude": True}},
+    )
+    repaired_content = _extract_chat_message_content(repair_data)
+    if not repaired_content:
+        raise LLMStructuredOutputError("V2 JSON repair returned empty content")
+    return repaired_content
+
+
+async def _score_subset_with_model_v2(
+    *,
+    settings: Any,
+    model_name: str,
+    articles: list[dict],
+    horizon_days: int,
+) -> tuple[dict[int, dict], list[int], int]:
+    """
+    Score subset with one model.
+
+    Returns:
+        (valid_results_by_id, failed_ids, parse_fail_count)
+    """
+    if not articles:
+        return {}, [], 0
+
+    expected_ids = [int(article["id"]) for article in articles]
+    user_prompt = _build_llm_v2_user_prompt(articles, horizon_days=horizon_days)
+
+    async def _request(strict_schema: bool, *, max_tokens: int) -> dict[str, Any]:
+        request_kwargs: dict[str, Any] = {
+            "api_key": settings.openrouter_api_key,
+            "model": model_name,
+            "messages": [
+                {"role": "system", "content": LLM_V2_SYSTEM_PROMPT},
+                {"role": "user", "content": user_prompt},
+            ],
+            "max_tokens": max_tokens,
+            "temperature": 0.0,
+            "timeout_seconds": 60.0,
+            "max_retries": settings.openrouter_max_retries,
+            "rpm": settings.openrouter_rpm,
+            "fallback_models": settings.openrouter_fallback_models_list,
+            "referer": "https://copper-mind.vercel.app",
+            "title": "CopperMind Sentiment Analysis V2",
+            "extra_payload": {"reasoning": {"exclude": True}},
+        }
+        if strict_schema:
+            request_kwargs["response_format"] = LLM_SCORING_RESPONSE_FORMAT_V2
+            request_kwargs["provider"] = LLM_SCORING_PROVIDER_OPTIONS
+        return await create_chat_completion(**request_kwargs)
+
+    async def _request_with_provider_fallback(*, max_tokens: int) -> dict[str, Any]:
+        try:
+            return await _request(strict_schema=True, max_tokens=max_tokens)
+        except OpenRouterError as exc:
+            message = str(exc).lower()
+            if exc.status_code == 404 and "no endpoints found" in message:
+                logger.warning(
+                    "V2 structured scoring unsupported by provider route for model=%s. Retrying relaxed mode.",
+                    model_name,
+                )
+                return await _request(strict_schema=False, max_tokens=max_tokens)
+            raise
+
+    parse_fail_count = 0
+    try:
+        data = await _request_with_provider_fallback(max_tokens=LLM_SCORING_MAX_TOKENS_PRIMARY)
+    except Exception:
+        return {}, expected_ids, len(expected_ids)
+
+    content = _extract_chat_message_content(data)
+    if not content:
+        finish_reason = _extract_finish_reason(data)
+        if finish_reason == "length":
+            data = await _request_with_provider_fallback(max_tokens=LLM_SCORING_MAX_TOKENS_RETRY)
+            content = _extract_chat_message_content(data)
+        if not content:
+            return {}, expected_ids, len(expected_ids)
+
+    try:
+        raw_results = json.loads(_clean_json_content(content))
+        valid, failed = _parse_llm_v2_items(
+            raw_results=raw_results,
+            expected_ids=expected_ids,
+            model_name=model_name,
+        )
+        parse_fail_count += len(failed)
+        return valid, failed, parse_fail_count
+    except Exception:
+        parse_fail_count += len(expected_ids)
+
+    try:
+        repaired = await _repair_json_response_v2(
+            settings=settings,
+            model_name=model_name,
+            malformed_content=content,
+            expected_ids=expected_ids,
+        )
+        raw_results = json.loads(_clean_json_content(repaired))
+        valid, failed = _parse_llm_v2_items(
+            raw_results=raw_results,
+            expected_ids=expected_ids,
+            model_name=model_name,
+        )
+        return valid, failed, parse_fail_count
+    except Exception:
+        return {}, expected_ids, parse_fail_count
+
+
+async def score_batch_with_llm_v2(
+    articles: list[dict],
+    *,
+    horizon_days: int = 5,
+) -> dict[str, Any]:
+    """
+    Commodity-aware sentiment scoring with fast+reliable model escalation.
+    """
+    settings = get_settings()
+    if not settings.openrouter_api_key:
+        raise RuntimeError("OpenRouter API key not configured")
+
+    fast_model = settings.resolved_scoring_fast_model
+    reliable_model = settings.resolved_scoring_reliable_model
+    conflict_threshold = _clip(
+        float(getattr(settings, "sentiment_escalate_conflict_threshold", 0.55)),
+        0.0,
+        1.0,
+    )
+
+    normalized_articles = [
+        {
+            "id": int(article["id"]),
+            "title": str(article.get("title") or "")[:500],
+            "description": str(article.get("description") or "")[:800],
+            "text": str(article.get("text") or "")[:1800],
+        }
+        for article in articles
+    ]
+    expected_ids = [item["id"] for item in normalized_articles]
+    article_by_id = {item["id"]: item for item in normalized_articles}
+
+    fast_valid, fast_failed, parse_fail_fast = await _score_subset_with_model_v2(
+        settings=settings,
+        model_name=fast_model,
+        articles=normalized_articles,
+        horizon_days=horizon_days,
+    )
+
+    results_by_id = dict(fast_valid)
+    parse_fail_total = int(parse_fail_fast)
+
+    conflict_ids: list[int] = []
+    for article_id, item in fast_valid.items():
+        event_type = _normalize_event_type(item.get("event_type"))
+        rule_sign = int(LLM_V2_EVENT_SIGN.get(event_type, 0))
+        llm_sign = _sign(float(item.get("impact_score", 0.0)))
+        conflict_strength = _clip(
+            float(item.get("confidence", 0.0)) * float(item.get("relevance", 0.0)),
+            0.0,
+            1.0,
+        )
+        if rule_sign != 0 and llm_sign != 0 and llm_sign != rule_sign and conflict_strength >= conflict_threshold:
+            conflict_ids.append(article_id)
+            results_by_id.pop(article_id, None)
+
+    escalation_ids = sorted(set(fast_failed).union(conflict_ids))
+    escalation_count = len(escalation_ids)
+
+    if escalation_ids:
+        reliable_subset = [
+            article_by_id[article_id]
+            for article_id in escalation_ids
+            if article_id in article_by_id
+        ]
+        reliable_valid, _reliable_failed, parse_fail_reliable = await _score_subset_with_model_v2(
+            settings=settings,
+            model_name=reliable_model,
+            articles=reliable_subset,
+            horizon_days=horizon_days,
+        )
+        results_by_id.update(reliable_valid)
+        parse_fail_total += int(parse_fail_reliable)
+
+    results = [results_by_id[article_id] for article_id in expected_ids if article_id in results_by_id]
+    failed_ids = [article_id for article_id in expected_ids if article_id not in results_by_id]
+    fallback_count = len(failed_ids)
+
+    return {
+        "results": results,
+        "failed_ids": failed_ids,
+        "parse_fail_count": parse_fail_total,
+        "escalation_count": escalation_count,
+        "fallback_count": fallback_count,
+        "model_fast": fast_model,
+        "model_reliable": reliable_model,
+    }
+
+
+def score_batch_with_finbert_v2(articles: list[dict]) -> dict[int, dict]:
+    """Score text with FinBERT for tone/intensity features."""
+    pipe = get_finbert_pipeline()
+    results: dict[int, dict] = {}
+
+    for article in articles:
+        article_id = int(article["id"])
+        text = str(
+            article.get("text")
+            or f"{article.get('title', '')} {article.get('description', '')}"
+        )[:1200]
+        scores = score_text_with_finbert(pipe, text)
+        results[article_id] = {
+            "prob_positive": float(scores["prob_positive"]),
+            "prob_neutral": float(scores["prob_neutral"]),
+            "prob_negative": float(scores["prob_negative"]),
+            "tone": float(scores["score"]),
+            "magnitude": abs(float(scores["prob_positive"]) - float(scores["prob_negative"])),
+        }
+
+    return results
+
+
+def compute_final_score_v2(
+    *,
+    impact_score_llm: float,
+    confidence_llm: float,
+    relevance_score: float,
+    event_type: str,
+    prob_positive: float,
+    prob_negative: float,
+) -> dict[str, float | int]:
+    """Compute deterministic ensemble score for V2."""
+    llm_impact = _clip(float(impact_score_llm), -1.0, 1.0)
+    llm_conf = _clip(float(confidence_llm), 0.0, 1.0)
+    relevance = _clip(float(relevance_score), 0.0, 1.0)
+    tone = _clip(float(prob_positive) - float(prob_negative), -1.0, 1.0)
+    tone_mag = abs(tone)
+
+    normalized_event = _normalize_event_type(event_type)
+    rule_sign = int(LLM_V2_EVENT_SIGN.get(normalized_event, 0))
+    rule_strength = float(LLM_V2_EVENT_STRENGTH.get(normalized_event, 0.25))
+
+    llm_sign = _sign(llm_impact)
+    final_sign = llm_sign if llm_sign != 0 else rule_sign
+    if final_sign == 0 and tone_mag >= 0.2:
+        final_sign = _sign(tone)
+
+    impact_mag = _clip(
+        (0.55 * abs(llm_impact))
+        + (0.25 * tone_mag)
+        + (0.20 * _clip(rule_strength, 0.0, 1.0)),
+        0.0,
+        1.0,
+    )
+    if final_sign == 0:
+        impact_mag = min(impact_mag, 0.12)
+
+    final_score = float(final_sign) * impact_mag
+    agreement = 1.0 if (rule_sign == 0 or llm_sign == 0 or llm_sign == rule_sign) else 0.4
+    confidence_cal = _clip((0.50 * llm_conf) + (0.30 * agreement) + (0.20 * relevance), 0.01, 0.99)
+
+    return {
+        "rule_sign": rule_sign,
+        "rule_strength": rule_strength,
+        "final_score": final_score,
+        "confidence_calibrated": confidence_cal,
+    }
+
+
+def _build_article_fallback_v2(
+    *,
+    article: dict,
+    finbert: dict,
+    model_fast: str,
+    model_reliable: str,
+) -> dict:
+    """Deterministic article-level fallback without zero-only outputs."""
+    text = str(article.get("text") or f"{article.get('title', '')} {article.get('description', '')}")
+    event_type = _infer_event_type_from_text(text)
+    rule_sign = int(LLM_V2_EVENT_SIGN.get(event_type, 0))
+    tone = float(finbert.get("tone", 0.0))
+    tone_sign = _sign(tone)
+    direction = rule_sign if rule_sign != 0 else tone_sign
+    if direction == 0:
+        impact_score = 0.0
+    else:
+        impact_score = float(direction) * _clip((abs(tone) * 0.35) + 0.08, 0.08, 0.25)
+
+    relevance = 0.10 if event_type == "non_copper" else 0.45
+    confidence = 0.18 if direction == 0 else _clip(0.22 + (abs(tone) * 0.22), 0.22, 0.45)
+
+    return {
+        "id": int(article["id"]),
+        "label": _label_from_impact_score(impact_score),
+        "impact_score": impact_score,
+        "confidence": confidence,
+        "relevance": relevance,
+        "event_type": event_type,
+        "reasoning": "deterministic_fallback",
+        "llm_model": model_fast,
+        "model_fast": model_fast,
+        "model_reliable": model_reliable,
+        "fallback_used": True,
+    }
+
+
+def score_unscored_processed_articles(
+    session: Session,
+    *,
+    chunk_size: int = 12,
+    backfill_days: Optional[int] = None,
+) -> dict[str, int]:
+    """
+    Score unscored `news_processed` articles into `news_sentiments_v2`.
+    """
+    settings = get_settings()
+    horizon_days = max(1, int(getattr(settings, "sentiment_horizon_days", 5)))
+    relevance_min = _clip(float(getattr(settings, "sentiment_relevance_min", 0.35)), 0.0, 1.0)
+
+    query = (
+        session.query(
+            NewsProcessed.id.label("processed_id"),
+            NewsProcessed.canonical_title,
+            NewsProcessed.cleaned_text,
+            NewsRaw.title.label("raw_title"),
+            NewsRaw.description.label("raw_description"),
+            NewsRaw.published_at,
+        )
+        .join(NewsRaw, NewsProcessed.raw_id == NewsRaw.id)
+        .outerjoin(
+            NewsSentimentV2,
+            (NewsProcessed.id == NewsSentimentV2.news_processed_id)
+            & (NewsSentimentV2.horizon_days == horizon_days),
+        )
+        .filter(NewsSentimentV2.id.is_(None))
+        .order_by(NewsRaw.published_at.asc(), NewsProcessed.id.asc())
+    )
+
+    if backfill_days is not None:
+        cutoff = datetime.now(timezone.utc) - timedelta(days=max(1, int(backfill_days)))
+        query = query.filter(NewsRaw.published_at >= cutoff)
+
+    rows = query.all()
+    if not rows:
+        logger.info("No unscored processed articles found")
+        return {
+            "scored_count": 0,
+            "parse_fail_count": 0,
+            "escalation_count": 0,
+            "fallback_count": 0,
+            "finbert_used": 0,
+        }
+
+    logger.info("Found %s unscored processed articles for V2 scoring", len(rows))
+
+    scored_count = 0
+    parse_fail_count = 0
+    escalation_count = 0
+    fallback_count = 0
+    finbert_used = 0
+    llm_budget_remaining = max(0, int(settings.max_llm_articles_per_run))
+    fast_model = settings.resolved_scoring_fast_model
+    reliable_model = settings.resolved_scoring_reliable_model
+
+    for chunk_idx in range(0, len(rows), chunk_size):
+        chunk_rows = rows[chunk_idx:chunk_idx + chunk_size]
+        chunk_items: list[dict] = []
+        for row in chunk_rows:
+            title = str(row.raw_title or row.canonical_title or "")[:500]
+            description = str(row.raw_description or "")[:1000]
+            text = str(row.cleaned_text or f"{title} {description}")[:2000]
+            chunk_items.append(
+                {
+                    "id": int(row.processed_id),
+                    "title": title,
+                    "description": description,
+                    "text": text,
+                    "published_at": row.published_at,
+                }
+            )
+
+        finbert_by_id = score_batch_with_finbert_v2(chunk_items)
+        finbert_used += len(finbert_by_id)
+
+        llm_results_by_id: dict[int, dict] = {}
+        llm_candidates: list[dict] = []
+        if settings.openrouter_api_key and llm_budget_remaining > 0:
+            llm_take = min(len(chunk_items), llm_budget_remaining)
+            llm_candidates = chunk_items[:llm_take]
+            llm_budget_remaining -= llm_take
+
+        if llm_candidates:
+            try:
+                llm_bundle = run_async_from_sync(
+                    score_batch_with_llm_v2,
+                    llm_candidates,
+                    horizon_days=horizon_days,
+                )
+                for item in llm_bundle.get("results", []):
+                    llm_results_by_id[int(item["id"])] = item
+                parse_fail_count += int(llm_bundle.get("parse_fail_count", 0))
+                escalation_count += int(llm_bundle.get("escalation_count", 0))
+                fast_model = str(llm_bundle.get("model_fast", fast_model))
+                reliable_model = str(llm_bundle.get("model_reliable", reliable_model))
+            except Exception as exc:
+                logger.warning("V2 LLM scoring failed for chunk starting at %s: %s", chunk_idx, exc)
+                parse_fail_count += len(llm_candidates)
+
+        for article in chunk_items:
+            article_id = int(article["id"])
+            finbert = finbert_by_id.get(article_id, _neutral_finbert_score())
+            llm = llm_results_by_id.get(article_id)
+
+            if llm is None:
+                llm = _build_article_fallback_v2(
+                    article=article,
+                    finbert=finbert if isinstance(finbert, dict) else {},
+                    model_fast=fast_model,
+                    model_reliable=reliable_model,
+                )
+            else:
+                llm["model_fast"] = fast_model
+                llm["model_reliable"] = reliable_model
+                llm["fallback_used"] = False
+
+            if bool(llm.get("fallback_used", False)):
+                fallback_count += 1
+
+            if float(llm.get("relevance", 0.0)) < relevance_min and llm.get("event_type") != "non_copper":
+                llm["event_type"] = "non_copper"
+                llm["label"] = "NEUTRAL"
+                llm["impact_score"] = 0.0
+
+            metrics = compute_final_score_v2(
+                impact_score_llm=float(llm.get("impact_score", 0.0)),
+                confidence_llm=float(llm.get("confidence", 0.01)),
+                relevance_score=float(llm.get("relevance", 0.01)),
+                event_type=str(llm.get("event_type", "mixed_unclear")),
+                prob_positive=float(finbert.get("prob_positive", 0.33)),
+                prob_negative=float(finbert.get("prob_negative", 0.33)),
+            )
+
+            payload = {
+                "label": llm.get("label", "NEUTRAL"),
+                "impact_score": round(float(llm.get("impact_score", 0.0)), 4),
+                "confidence": round(float(llm.get("confidence", 0.01)), 4),
+                "relevance": round(float(llm.get("relevance", 0.01)), 4),
+                "event_type": llm.get("event_type", "mixed_unclear"),
+                "reasoning": llm.get("reasoning", ""),
+                "rule_sign": metrics["rule_sign"],
+                "rule_strength": round(float(metrics["rule_strength"]), 4),
+                "confidence_calibrated": round(float(metrics["confidence_calibrated"]), 4),
+                "fallback_used": bool(llm.get("fallback_used", False)),
+                "llm_model": llm.get("llm_model", fast_model),
+                "scoring_version": SCORING_V2_VERSION,
+            }
+
+            sentiment_v2 = NewsSentimentV2(
+                news_processed_id=article_id,
+                horizon_days=horizon_days,
+                label=str(llm.get("label", "NEUTRAL")),
+                impact_score_llm=float(llm.get("impact_score", 0.0)),
+                confidence_llm=float(llm.get("confidence", 0.01)),
+                confidence_calibrated=float(metrics["confidence_calibrated"]),
+                relevance_score=float(llm.get("relevance", 0.01)),
+                event_type=str(llm.get("event_type", "mixed_unclear")),
+                rule_sign=int(metrics["rule_sign"]),
+                final_score=float(metrics["final_score"]),
+                finbert_pos=float(finbert.get("prob_positive", 0.33)),
+                finbert_neu=float(finbert.get("prob_neutral", 0.34)),
+                finbert_neg=float(finbert.get("prob_negative", 0.33)),
+                reasoning_json=json.dumps(payload, ensure_ascii=True),
+                model_fast=fast_model,
+                model_reliable=reliable_model,
+                scored_at=datetime.now(timezone.utc),
+            )
+            session.add(sentiment_v2)
+            scored_count += 1
+
+        session.commit()
+
+    logger.info(
+        "V2 scoring summary: scored=%s parse_fail=%s escalations=%s fallback=%s finbert_used=%s",
+        scored_count,
+        parse_fail_count,
+        escalation_count,
+        fallback_count,
+        finbert_used,
+    )
+    return {
+        "scored_count": scored_count,
+        "parse_fail_count": parse_fail_count,
+        "escalation_count": escalation_count,
+        "fallback_count": fallback_count,
+        "finbert_used": finbert_used,
+    }
+
+
+def aggregate_daily_sentiment_v2(
+    session: Session,
+    *,
+    tau_hours: float = 12.0,
+) -> int:
+    """Aggregate V2 article scores into daily_sentiments_v2."""
+    settings = get_settings()
+    tau_hours = tau_hours or settings.sentiment_tau_hours
+    horizon_days = max(1, int(getattr(settings, "sentiment_horizon_days", 5)))
+    relevance_min = _clip(float(getattr(settings, "sentiment_relevance_min", 0.35)), 0.0, 1.0)
+
+    rows = (
+        session.query(
+            NewsRaw.published_at,
+            NewsSentimentV2.final_score,
+            NewsSentimentV2.confidence_calibrated,
+            NewsSentimentV2.relevance_score,
+        )
+        .join(NewsProcessed, NewsProcessed.raw_id == NewsRaw.id)
+        .join(
+            NewsSentimentV2,
+            (NewsSentimentV2.news_processed_id == NewsProcessed.id)
+            & (NewsSentimentV2.horizon_days == horizon_days),
+        )
+        .filter(NewsSentimentV2.relevance_score >= relevance_min)
+        .all()
+    )
+
+    if not rows:
+        logger.info("No V2 scored articles available for daily aggregation")
+        return 0
+
+    df = pd.DataFrame(
+        rows,
+        columns=["published_at", "final_score", "confidence_calibrated", "relevance_score"],
+    )
+    df["date"] = pd.to_datetime(df["published_at"]).dt.normalize()
+
+    def calc_weights(group):
+        hours = (group["published_at"] - group["date"]).dt.total_seconds() / 3600.0
+        weights = np.exp(hours / tau_hours)
+        return weights / weights.sum()
+
+    daily_rows = []
+    for date, group in df.groupby("date"):
+        weights = calc_weights(group)
+        daily_rows.append(
+            {
+                "date": date,
+                "sentiment_index": float((group["final_score"] * weights).sum()),
+                "news_count": int(len(group)),
+                "avg_confidence": float(group["confidence_calibrated"].mean()),
+                "avg_relevance": float(group["relevance_score"].mean()),
+            }
+        )
+
+    count = 0
+    for row in daily_rows:
+        date_dt = row["date"].to_pydatetime()
+        if date_dt.tzinfo is None:
+            date_dt = date_dt.replace(tzinfo=timezone.utc)
+
+        existing = session.query(DailySentimentV2).filter(
+            func.date(DailySentimentV2.date) == func.date(date_dt)
+        ).first()
+        if existing:
+            existing.sentiment_index = row["sentiment_index"]
+            existing.news_count = row["news_count"]
+            existing.avg_confidence = row["avg_confidence"]
+            existing.avg_relevance = row["avg_relevance"]
+            existing.source_version = "v2"
+            existing.aggregated_at = datetime.now(timezone.utc)
+        else:
+            session.add(
+                DailySentimentV2(
+                    date=date_dt,
+                    sentiment_index=row["sentiment_index"],
+                    news_count=row["news_count"],
+                    avg_confidence=row["avg_confidence"],
+                    avg_relevance=row["avg_relevance"],
+                    source_version="v2",
+                    aggregated_at=datetime.now(timezone.utc),
+                )
+            )
+        count += 1
+
+    session.commit()
+    logger.info("Aggregated V2 sentiment for %s days", count)
+    return count
+
+
+def backfill_sentiment_v2(
+    session: Session,
+    *,
+    days: int = 180,
+    batch_size: int = 50,
+) -> dict[str, int]:
+    """Idempotent V2 backfill helper for last N days."""
+    logger.info("Starting V2 backfill for last %s days (batch_size=%s)", days, batch_size)
+    return score_unscored_processed_articles(
+        session=session,
+        chunk_size=batch_size,
+        backfill_days=days,
+    )
+
+
 def score_unscored_articles(
     session: Session,
     chunk_size: int = 12
@@ -1374,18 +2324,31 @@ def run_full_pipeline(
     Returns:
         Dict with results from each stage
     """
+    settings = get_settings()
     results = {
         "scored_articles": 0,
+        "scored_articles_v2": 0,
         "aggregated_days": 0,
+        "aggregated_days_v2": 0,
         "model_result": None,
         "timestamp": datetime.now(timezone.utc).isoformat(),
     }
     
     with SessionLocal() as session:
         if score_sentiment:
-            results["scored_articles"] = score_unscored_articles(session)
+            if settings.scoring_source == "news_processed":
+                scoring_stats = score_unscored_processed_articles(session)
+                results["scored_articles"] = int(scoring_stats.get("scored_count", 0))
+                results["scored_articles_v2"] = int(scoring_stats.get("scored_count", 0))
+                results["llm_parse_fail_count"] = int(scoring_stats.get("parse_fail_count", 0))
+                results["escalation_count"] = int(scoring_stats.get("escalation_count", 0))
+                results["fallback_count"] = int(scoring_stats.get("fallback_count", 0))
+            else:
+                results["scored_articles"] = score_unscored_articles(session)
         
         if aggregate_sentiment:
+            if settings.scoring_source == "news_processed":
+                results["aggregated_days_v2"] = aggregate_daily_sentiment_v2(session)
             results["aggregated_days"] = aggregate_daily_sentiment(session)
         
         if train_model:
@@ -1426,6 +2389,18 @@ def main():
         action="store_true",
         help="Run sentiment scoring + daily aggregation (no training)"
     )
+    parser.add_argument(
+        "--backfill-v2-days",
+        type=int,
+        default=0,
+        help="Backfill unscored V2 sentiment for last N days (idempotent)"
+    )
+    parser.add_argument(
+        "--backfill-v2-batch-size",
+        type=int,
+        default=50,
+        help="Batch size for V2 backfill mode"
+    )
     parser.add_argument(
         "--target-symbol",
         type=str,
@@ -1452,8 +2427,9 @@ def main():
     score = args.run_all or args.score_only or args.refresh_sentiment
     aggregate = args.run_all or args.aggregate_only or args.refresh_sentiment
     train = args.run_all or args.train_only
+    backfill_v2 = args.backfill_v2_days > 0
     
-    if not (score or aggregate or train):
+    if not (score or aggregate or train or backfill_v2):
         parser.print_help()
         return
     
@@ -1463,6 +2439,26 @@ def main():
     
     # Run pipeline
     def do_run():
+        if backfill_v2:
+            with SessionLocal() as session:
+                scoring_stats = backfill_sentiment_v2(
+                    session,
+                    days=args.backfill_v2_days,
+                    batch_size=max(1, int(args.backfill_v2_batch_size)),
+                )
+                aggregated_days_v2 = aggregate_daily_sentiment_v2(session)
+                return {
+                    "timestamp": datetime.now(timezone.utc).isoformat(),
+                    "scored_articles": int(scoring_stats.get("scored_count", 0)),
+                    "scored_articles_v2": int(scoring_stats.get("scored_count", 0)),
+                    "aggregated_days": 0,
+                    "aggregated_days_v2": int(aggregated_days_v2),
+                    "llm_parse_fail_count": int(scoring_stats.get("parse_fail_count", 0)),
+                    "escalation_count": int(scoring_stats.get("escalation_count", 0)),
+                    "fallback_count": int(scoring_stats.get("fallback_count", 0)),
+                    "model_result": None,
+                    "backfill_days": int(args.backfill_v2_days),
+                }
         return run_full_pipeline(
             target_symbol=args.target_symbol,
             score_sentiment=score,
@@ -1487,9 +2483,21 @@ def main():
     
     if score:
         print(f"\nSentiment Scoring: {results['scored_articles']} articles")
+        if "scored_articles_v2" in results:
+            print(f"V2 Sentiment Scoring: {results.get('scored_articles_v2', 0)} articles")
+        if "llm_parse_fail_count" in results:
+            print(f"  - LLM parse failures: {results.get('llm_parse_fail_count', 0)}")
+            print(f"  - Escalations: {results.get('escalation_count', 0)}")
+            print(f"  - Deterministic fallbacks: {results.get('fallback_count', 0)}")
     
     if aggregate:
         print(f"Daily Aggregation: {results['aggregated_days']} days")
+        if results.get("aggregated_days_v2") is not None:
+            print(f"Daily Aggregation V2 (shadow): {results.get('aggregated_days_v2', 0)} days")
+
+    if backfill_v2:
+        print(f"\nBackfill V2 Days: {results.get('backfill_days', 0)}")
+        print(f"V2 Aggregation Days: {results.get('aggregated_days_v2', 0)}")
     
     if train and results.get("model_result"):
         mr = results["model_result"]

app/db.py

@@ -148,6 +148,38 @@ def _run_migrations(engine):
         except Exception as e:
             logger.debug(f"Migration check for ai_stance: {e}")
 
+        # Migration: V2 stage-2 metrics columns
+        v2_metric_columns = [
+            ("articles_scored_v2", "INTEGER"),
+            ("llm_parse_fail_count", "INTEGER"),
+            ("escalation_count", "INTEGER"),
+            ("fallback_count", "INTEGER"),
+        ]
+        try:
+            if is_sqlite:
+                result = conn.execute(text("PRAGMA table_info(pipeline_run_metrics)"))
+                columns = [row[1] for row in result.fetchall()]
+                for column_name, column_type in v2_metric_columns:
+                    if column_name not in columns:
+                        conn.execute(
+                            text(
+                                f"ALTER TABLE pipeline_run_metrics ADD COLUMN {column_name} {column_type}"
+                            )
+                        )
+                conn.commit()
+            else:
+                for column_name, column_type in v2_metric_columns:
+                    conn.execute(
+                        text(
+                            f"ALTER TABLE pipeline_run_metrics "
+                            f"ADD COLUMN IF NOT EXISTS {column_name} {column_type}"
+                        )
+                    )
+                conn.commit()
+            logger.info("Migration: Ensured V2 Stage-2 metric columns exist")
+        except Exception as e:
+            logger.debug(f"Migration check for V2 metric columns: {e}")
+
 
 def init_db():
     """

app/features.py

@@ -19,7 +19,7 @@ from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 from app.db import SessionLocal
-from app.models import PriceBar, DailySentiment
+from app.models import PriceBar, DailySentiment, DailySentimentV2
 from app.settings import get_settings
 
 logger = logging.getLogger(__name__)
@@ -81,21 +81,37 @@ def load_sentiment_data(
     Returns:
         DataFrame with columns: date, sentiment_index, news_count
     """
-    query = session.query(
-        DailySentiment.date,
-        DailySentiment.sentiment_index,
-        DailySentiment.news_count
-    )
-    
-    if start_date:
-        query = query.filter(DailySentiment.date >= start_date)
-    if end_date:
-        query = query.filter(DailySentiment.date <= end_date)
-    
-    query = query.order_by(DailySentiment.date.asc())
-    
-    rows = query.all()
-    
+    settings = get_settings()
+    source = str(getattr(settings, "scoring_source", "news_articles")).strip().lower()
+    use_v2 = source == "news_processed"
+
+    rows = []
+    if use_v2:
+        query_v2 = session.query(
+            DailySentimentV2.date,
+            DailySentimentV2.sentiment_index,
+            DailySentimentV2.news_count
+        )
+        if start_date:
+            query_v2 = query_v2.filter(DailySentimentV2.date >= start_date)
+        if end_date:
+            query_v2 = query_v2.filter(DailySentimentV2.date <= end_date)
+        rows = query_v2.order_by(DailySentimentV2.date.asc()).all()
+        if not rows:
+            logger.warning("No rows in daily_sentiments_v2; falling back to daily_sentiments")
+
+    if not rows:
+        query = session.query(
+            DailySentiment.date,
+            DailySentiment.sentiment_index,
+            DailySentiment.news_count
+        )
+        if start_date:
+            query = query.filter(DailySentiment.date >= start_date)
+        if end_date:
+            query = query.filter(DailySentiment.date <= end_date)
+        rows = query.order_by(DailySentiment.date.asc()).all()
+
     if not rows:
         return pd.DataFrame()
     

app/inference.py

@@ -25,7 +25,17 @@ from sqlalchemy import func
 from sqlalchemy.orm import Session
 
 from app.db import SessionLocal
-from app.models import PriceBar, DailySentiment, AnalysisSnapshot, NewsArticle, NewsSentiment
+from app.models import (
+    PriceBar,
+    DailySentiment,
+    DailySentimentV2,
+    AnalysisSnapshot,
+    NewsArticle,
+    NewsSentiment,
+    NewsRaw,
+    NewsProcessed,
+    NewsSentimentV2,
+)
 from app.settings import get_settings
 from app.features import (
     load_price_data,
@@ -167,10 +177,20 @@ def get_current_price(session: Session, symbol: str) -> Optional[float]:
 
 def get_current_sentiment(session: Session) -> Optional[float]:
     """Get the most recent daily sentiment index."""
+    settings = get_settings()
+    source = str(getattr(settings, "scoring_source", "news_articles")).strip().lower()
+
+    if source == "news_processed":
+        latest_v2 = session.query(DailySentimentV2).order_by(
+            DailySentimentV2.date.desc()
+        ).first()
+        if latest_v2 is not None:
+            return latest_v2.sentiment_index
+        logger.warning("No rows in daily_sentiments_v2; falling back to legacy daily_sentiments")
+
     latest = session.query(DailySentiment).order_by(
         DailySentiment.date.desc()
     ).first()
-    
     return latest.sentiment_index if latest else None
 
 
@@ -180,18 +200,38 @@ def get_data_quality_stats(
     days: int = 7
 ) -> dict:
     """Get data quality statistics for the report."""
+    settings = get_settings()
+    source = str(getattr(settings, "scoring_source", "news_articles")).strip().lower()
     cutoff = datetime.now(timezone.utc) - timedelta(days=days)
-    
-    # News count
-    news_count = session.query(func.count(NewsArticle.id)).filter(
-        NewsArticle.published_at >= cutoff
-    ).scalar()
-    
-    # Scored news count
-    scored_count = session.query(func.count(NewsSentiment.id)).join(
-        NewsArticle,
-        NewsSentiment.news_article_id == NewsArticle.id
-    ).filter(NewsArticle.published_at >= cutoff).scalar()
+
+    if source == "news_processed":
+        horizon_days = max(1, int(getattr(settings, "sentiment_horizon_days", 5)))
+        news_count = session.query(func.count(NewsProcessed.id)).join(
+            NewsRaw, NewsProcessed.raw_id == NewsRaw.id
+        ).filter(
+            NewsRaw.published_at >= cutoff
+        ).scalar()
+
+        scored_count = session.query(func.count(NewsSentimentV2.id)).join(
+            NewsProcessed,
+            NewsSentimentV2.news_processed_id == NewsProcessed.id
+        ).join(
+            NewsRaw,
+            NewsProcessed.raw_id == NewsRaw.id
+        ).filter(
+            NewsRaw.published_at >= cutoff,
+            NewsSentimentV2.horizon_days == horizon_days,
+        ).scalar()
+    else:
+        # Legacy article-level stats
+        news_count = session.query(func.count(NewsArticle.id)).filter(
+            NewsArticle.published_at >= cutoff
+        ).scalar()
+
+        scored_count = session.query(func.count(NewsSentiment.id)).join(
+            NewsArticle,
+            NewsSentiment.news_article_id == NewsArticle.id
+        ).filter(NewsArticle.published_at >= cutoff).scalar()
     
     # Price bar coverage
     expected_days = days

app/main.py

@@ -22,7 +22,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from sqlalchemy import func
 
 from app.db import init_db, SessionLocal, get_db_type
-from app.models import NewsArticle, PriceBar, DailySentiment, AnalysisSnapshot
+from app.models import NewsArticle, PriceBar, DailySentiment, DailySentimentV2, AnalysisSnapshot
 from app.settings import get_settings
 from app.lock import is_pipeline_locked
 # NOTE: Faz 1 - API is snapshot-only, no report generation
@@ -219,6 +219,9 @@ async def get_history(
     IMPORTANT: sentiment_index of 0.0 is a valid value (neutral sentiment),
     not the same as missing data. We return explicit 0.0 values.
     """
+    settings = get_settings()
+    source = str(getattr(settings, "scoring_source", "news_articles")).strip().lower()
+
     with SessionLocal() as session:
         # Calculate date range
         end_date = datetime.now(timezone.utc)
@@ -239,14 +242,28 @@ async def get_history(
                 detail=f"No price data found for {symbol}"
             )
         
-        # Query sentiment
-        sentiments = session.query(
-            DailySentiment.date,
-            DailySentiment.sentiment_index,
-            DailySentiment.news_count
-        ).filter(
-            DailySentiment.date >= start_date
-        ).order_by(DailySentiment.date.asc()).all()
+        # Query sentiment (prefer V2 when scoring source is news_processed)
+        sentiments = []
+        if source == "news_processed":
+            sentiments = session.query(
+                DailySentimentV2.date,
+                DailySentimentV2.sentiment_index,
+                DailySentimentV2.news_count
+            ).filter(
+                DailySentimentV2.date >= start_date
+            ).order_by(DailySentimentV2.date.asc()).all()
+
+            if not sentiments:
+                logger.warning("No rows in daily_sentiments_v2 for history; falling back to daily_sentiments")
+
+        if not sentiments:
+            sentiments = session.query(
+                DailySentiment.date,
+                DailySentiment.sentiment_index,
+                DailySentiment.news_count
+            ).filter(
+                DailySentiment.date >= start_date
+            ).order_by(DailySentiment.date.asc()).all()
         
         # Create sentiment lookup (by date string for easy matching)
         sentiment_lookup = {}

app/models.py

@@ -316,6 +316,10 @@ class PipelineRunMetrics(Base):
     news_raw_duplicates = Column(Integer, nullable=True)
     news_processed_inserted = Column(Integer, nullable=True)
     news_processed_duplicates = Column(Integer, nullable=True)
+    articles_scored_v2 = Column(Integer, nullable=True)
+    llm_parse_fail_count = Column(Integer, nullable=True)
+    escalation_count = Column(Integer, nullable=True)
+    fallback_count = Column(Integer, nullable=True)
     
     # Snapshot info
     snapshot_generated = Column(Boolean, default=False)
@@ -424,6 +428,80 @@ class NewsProcessed(Base):
     
     # Relationship
     raw = relationship("NewsRaw", back_populates="processed_items")
+    sentiment_v2_items = relationship("NewsSentimentV2", back_populates="processed")
     
     def __repr__(self):
         return f"<NewsProcessed(id={self.id}, dedup_key='{self.dedup_key[:16]}...')>"
+
+
+class NewsSentimentV2(Base):
+    """
+    Commodity-aware sentiment scores generated from news_processed records.
+    """
+
+    __tablename__ = "news_sentiments_v2"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+
+    news_processed_id = Column(
+        BigInteger,
+        ForeignKey("news_processed.id", ondelete="CASCADE"),
+        nullable=False,
+        index=True,
+    )
+    horizon_days = Column(Integer, nullable=False, default=5)
+
+    label = Column(String(20), nullable=False, index=True)
+    impact_score_llm = Column(Float, nullable=False)
+    confidence_llm = Column(Float, nullable=False)
+    confidence_calibrated = Column(Float, nullable=False, index=True)
+    relevance_score = Column(Float, nullable=False, index=True)
+    event_type = Column(String(50), nullable=False, index=True)
+    rule_sign = Column(Integer, nullable=False)
+    final_score = Column(Float, nullable=False, index=True)
+
+    finbert_pos = Column(Float, nullable=False)
+    finbert_neu = Column(Float, nullable=False)
+    finbert_neg = Column(Float, nullable=False)
+
+    reasoning_json = Column(Text, nullable=True)
+    model_fast = Column(String(100), nullable=True)
+    model_reliable = Column(String(100), nullable=True)
+    scored_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow, index=True)
+
+    processed = relationship("NewsProcessed", back_populates="sentiment_v2_items")
+
+    __table_args__ = (
+        UniqueConstraint("news_processed_id", "horizon_days", name="uq_news_sentiments_v2_processed_horizon"),
+        Index("ix_news_sentiments_v2_processed_scored", "news_processed_id", "scored_at"),
+    )
+
+    def __repr__(self):
+        return (
+            "<NewsSentimentV2(processed_id="
+            f"{self.news_processed_id}, horizon_days={self.horizon_days}, final_score={self.final_score:.3f})>"
+        )
+
+
+class DailySentimentV2(Base):
+    """
+    Daily aggregate sentiment generated from NewsSentimentV2.
+    """
+
+    __tablename__ = "daily_sentiments_v2"
+
+    id = Column(BigInteger, primary_key=True, autoincrement=True)
+    date = Column(DateTime(timezone=True), nullable=False, unique=True, index=True)
+
+    sentiment_index = Column(Float, nullable=False, index=True)
+    news_count = Column(Integer, nullable=False, default=0)
+    avg_confidence = Column(Float, nullable=True)
+    avg_relevance = Column(Float, nullable=True)
+    source_version = Column(String(20), nullable=False, default="v2")
+    aggregated_at = Column(DateTime(timezone=True), nullable=False, default=datetime.utcnow, index=True)
+
+    def __repr__(self):
+        return (
+            "<DailySentimentV2(date="
+            f"{self.date}, sentiment_index={self.sentiment_index:.3f}, news_count={self.news_count})>"
+        )

app/settings.py

@@ -53,6 +53,10 @@ class Settings(BaseSettings):
     sentiment_soft_neutral_polarity_threshold: float = 0.12
     sentiment_soft_neutral_max_mag: float = 0.25
     sentiment_soft_neutral_scale: float = 0.8
+    sentiment_relevance_min: float = 0.35
+    sentiment_escalate_conflict_threshold: float = 0.55
+    sentiment_horizon_days: int = 5
+    scoring_source: str = "news_processed"
     
     # API settings
     analysis_ttl_minutes: int = 30
@@ -73,10 +77,12 @@ class Settings(BaseSettings):
     # OpenRouter AI Commentary
     openrouter_api_key: Optional[str] = None
     # Deprecated - kept for backward compatibility
-    openrouter_model: str = "openai/gpt-oss-120b:free"
+    openrouter_model: str = "arcee-ai/trinity-large-preview:free"
     # New primary config
-    openrouter_model_scoring: str = "stepfun/step-3.5-flash:free"
-    openrouter_model_commentary: str = "stepfun/step-3.5-flash:free"
+    openrouter_model_scoring: str = "arcee-ai/trinity-large-preview:free"
+    openrouter_model_scoring_fast: Optional[str] = None
+    openrouter_model_scoring_reliable: Optional[str] = None
+    openrouter_model_commentary: str = "arcee-ai/trinity-large-preview:free"
     openrouter_rpm: int = 18
     openrouter_max_retries: int = 3
     max_llm_articles_per_run: int = 200
@@ -97,7 +103,7 @@ class Settings(BaseSettings):
     
     # LLM Sentiment Analysis
     # Deprecated - kept for backward compatibility
-    llm_sentiment_model: str = "openai/gpt-oss-120b:free"
+    llm_sentiment_model: str = "arcee-ai/trinity-large-preview:free"
     
     # Pipeline trigger authentication
     pipeline_trigger_secret: Optional[str] = None
@@ -194,11 +200,30 @@ class Settings(BaseSettings):
         """Preferred scoring model with backward-compatible fallback chain."""
         return (
             self._first_non_empty(
+                self.openrouter_model_scoring_fast,
                 self.openrouter_model_scoring,
                 self.llm_sentiment_model,
                 self.openrouter_model,
             )
-            or "stepfun/step-3.5-flash:free"
+            or "arcee-ai/trinity-large-preview:free"
+        )
+
+    @property
+    def resolved_scoring_fast_model(self) -> str:
+        """Fast model used for primary sentiment scoring."""
+        return self.resolved_scoring_model
+
+    @property
+    def resolved_scoring_reliable_model(self) -> str:
+        """Reliable model used for escalation/retry on malformed outputs."""
+        return (
+            self._first_non_empty(
+                self.openrouter_model_scoring_reliable,
+                self.openrouter_model,
+                self.llm_sentiment_model,
+                self.openrouter_model_scoring,
+            )
+            or "arcee-ai/trinity-large-preview:free"
         )
 
     @property
@@ -210,7 +235,7 @@ class Settings(BaseSettings):
                 self.openrouter_model,
                 self.llm_sentiment_model,
             )
-            or "stepfun/step-3.5-flash:free"
+            or "arcee-ai/trinity-large-preview:free"
         )
 
     @property

app/stepfun_probe.py

@@ -0,0 +1,237 @@
+"""
+StepFun / OpenRouter probe utility for sentiment V2 diagnostics.
+
+Usage:
+    py -m app.stepfun_probe
+    py -m app.stepfun_probe --sample-size 24 --db-limit 24
+    py -m app.stepfun_probe --skip-db
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import time
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.ai_engine import (
+    LLM_SCORING_PROVIDER_OPTIONS,
+    LLM_SCORING_RESPONSE_FORMAT_V2,
+    _build_llm_v2_user_prompt,
+    score_batch_with_llm_v2,
+)
+from app.db import SessionLocal, init_db
+from app.models import NewsProcessed, NewsRaw
+from app.openrouter_client import OpenRouterError, create_chat_completion
+from app.settings import get_settings
+
+
+def _build_handcrafted_articles() -> list[dict[str, Any]]:
+    return [
+        {
+            "id": 1,
+            "title": "Major copper mine outage in Chile removes 180k tonnes from expected supply",
+            "description": "Analysts expect lower exchange inventories and tighter concentrates market.",
+            "text": (
+                "Major copper mine outage in Chile removes 180k tonnes from expected supply. "
+                "Analysts expect lower exchange inventories and tighter concentrates market."
+            ),
+        },
+        {
+            "id": 2,
+            "title": "China property slowdown deepens and cable demand weakens",
+            "description": "Fabricators report softer orders and lower cathode premiums.",
+            "text": (
+                "China property slowdown deepens and cable demand weakens. "
+                "Fabricators report softer orders and lower cathode premiums."
+            ),
+        },
+        {
+            "id": 3,
+            "title": "Semiconductor patent lawsuit update",
+            "description": "No direct discussion of copper demand or supply.",
+            "text": "Semiconductor patent lawsuit update with no direct copper market linkage.",
+        },
+    ]
+
+
+def _load_db_articles(session: Session, limit: int) -> list[dict[str, Any]]:
+    rows = (
+        session.query(
+            NewsProcessed.id.label("processed_id"),
+            NewsRaw.title,
+            NewsRaw.description,
+            NewsProcessed.cleaned_text,
+            NewsRaw.published_at,
+        )
+        .join(NewsRaw, NewsProcessed.raw_id == NewsRaw.id)
+        .order_by(NewsRaw.published_at.desc(), NewsProcessed.id.desc())
+        .limit(max(1, int(limit)))
+        .all()
+    )
+
+    articles: list[dict[str, Any]] = []
+    for row in rows:
+        title = str(row.title or "")[:500]
+        description = str(row.description or "")[:800]
+        text = str(row.cleaned_text or f"{title} {description}")[:1800]
+        articles.append(
+            {
+                "id": int(row.processed_id),
+                "title": title,
+                "description": description,
+                "text": text,
+            }
+        )
+    return articles
+
+
+async def _run_strict_probe(
+    *,
+    model: str,
+    articles: list[dict[str, Any]],
+) -> dict[str, Any]:
+    settings = get_settings()
+    user_prompt = _build_llm_v2_user_prompt(articles, horizon_days=5)
+    started = time.perf_counter()
+    try:
+        data = await create_chat_completion(
+            api_key=settings.openrouter_api_key or "",
+            model=model,
+            messages=[
+                {
+                    "role": "system",
+                    "content": (
+                        "You are a Senior Copper Futures Analyst. Return only valid JSON array with keys: "
+                        "id,label,impact_score,confidence,relevance,event_type,reasoning."
+                    ),
+                },
+                {"role": "user", "content": user_prompt},
+            ],
+            max_tokens=1800,
+            temperature=0.0,
+            timeout_seconds=60.0,
+            max_retries=settings.openrouter_max_retries,
+            rpm=settings.openrouter_rpm,
+            fallback_models=settings.openrouter_fallback_models_list,
+            response_format=LLM_SCORING_RESPONSE_FORMAT_V2,
+            provider=LLM_SCORING_PROVIDER_OPTIONS,
+            extra_payload={"reasoning": {"exclude": True}},
+        )
+        elapsed = time.perf_counter() - started
+        choice = data.get("choices", [{}])[0]
+        content = choice.get("message", {}).get("content")
+        return {
+            "ok": True,
+            "elapsed_sec": round(elapsed, 2),
+            "finish_reason": choice.get("finish_reason"),
+            "content_len": len(content or ""),
+            "preview": str(content)[:240],
+        }
+    except OpenRouterError as exc:
+        elapsed = time.perf_counter() - started
+        return {
+            "ok": False,
+            "elapsed_sec": round(elapsed, 2),
+            "error_type": "OpenRouterError",
+            "status_code": exc.status_code,
+            "message": str(exc)[:280],
+        }
+    except Exception as exc:  # noqa: BLE001
+        elapsed = time.perf_counter() - started
+        return {
+            "ok": False,
+            "elapsed_sec": round(elapsed, 2),
+            "error_type": type(exc).__name__,
+            "message": str(exc)[:280],
+        }
+
+
+async def _run_v2_probe(
+    *,
+    articles: list[dict[str, Any]],
+    horizon_days: int,
+) -> dict[str, Any]:
+    started = time.perf_counter()
+    bundle = await score_batch_with_llm_v2(articles, horizon_days=horizon_days)
+    elapsed = time.perf_counter() - started
+
+    sample_results = []
+    for item in bundle.get("results", [])[:3]:
+        sample_results.append(
+            {
+                "id": item.get("id"),
+                "label": item.get("label"),
+                "impact_score": item.get("impact_score"),
+                "confidence": item.get("confidence"),
+                "relevance": item.get("relevance"),
+                "event_type": item.get("event_type"),
+            }
+        )
+
+    return {
+        "elapsed_sec": round(elapsed, 2),
+        "result_count": len(bundle.get("results", [])),
+        "fallback_count": int(bundle.get("fallback_count", 0)),
+        "parse_fail_count": int(bundle.get("parse_fail_count", 0)),
+        "escalation_count": int(bundle.get("escalation_count", 0)),
+        "failed_ids": bundle.get("failed_ids", []),
+        "model_fast": bundle.get("model_fast"),
+        "model_reliable": bundle.get("model_reliable"),
+        "sample_results": sample_results,
+    }
+
+
+async def _run_probe(sample_size: int, db_limit: int, skip_db: bool) -> None:
+    settings = get_settings()
+    fast_model = settings.resolved_scoring_fast_model
+    reliable_model = settings.resolved_scoring_reliable_model
+
+    print("=== StepFun Probe ===")
+    print(f"fast_model={fast_model}")
+    print(f"reliable_model={reliable_model}")
+    print(f"openrouter_rpm={settings.openrouter_rpm} max_retries={settings.openrouter_max_retries}")
+
+    handcrafted = _build_handcrafted_articles()
+
+    strict_summary = await _run_strict_probe(model=fast_model, articles=handcrafted)
+    print("\n[1] strict_schema_probe")
+    print(json.dumps(strict_summary, ensure_ascii=True, indent=2))
+
+    v2_smoke = await _run_v2_probe(articles=handcrafted, horizon_days=5)
+    print("\n[2] v2_smoke_probe_handcrafted")
+    print(json.dumps(v2_smoke, ensure_ascii=True, indent=2))
+
+    if skip_db:
+        return
+
+    with SessionLocal() as session:
+        db_articles = _load_db_articles(session, limit=db_limit)
+    if not db_articles:
+        print("\n[3] v2_db_probe: no articles found in news_processed")
+        return
+
+    db_probe_articles = db_articles[: max(1, min(sample_size, len(db_articles)))]
+    v2_db = await _run_v2_probe(articles=db_probe_articles, horizon_days=5)
+    print("\n[3] v2_db_probe")
+    print(f"sampled_articles={len(db_probe_articles)}")
+    print(json.dumps(v2_db, ensure_ascii=True, indent=2))
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Probe StepFun/OpenRouter behavior for sentiment V2.")
+    parser.add_argument("--sample-size", type=int, default=12, help="How many DB articles to score in DB probe.")
+    parser.add_argument("--db-limit", type=int, default=24, help="How many latest DB articles to fetch before sampling.")
+    parser.add_argument("--skip-db", action="store_true", help="Skip DB probe; run only strict + handcrafted tests.")
+    args = parser.parse_args()
+
+    init_db()
+    asyncio.run(_run_probe(sample_size=args.sample_size, db_limit=args.db_limit, skip_db=args.skip_db))
+
+
+if __name__ == "__main__":
+    main()
+

migrations/002_sentiment_v2.sql

@@ -0,0 +1,76 @@
+-- Migration: 002_sentiment_v2.sql
+-- Commodity-aware sentiment pipeline (V2)
+--
+-- Adds:
+--   1. news_sentiments_v2 (article-level scores on news_processed)
+--   2. daily_sentiments_v2 (daily aggregate from V2 scores)
+--   3. Stage-2 V2 run metrics columns on pipeline_run_metrics
+
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS news_sentiments_v2 (
+    id BIGSERIAL PRIMARY KEY,
+    news_processed_id BIGINT NOT NULL REFERENCES news_processed(id) ON DELETE CASCADE,
+    horizon_days INTEGER NOT NULL DEFAULT 5,
+
+    label VARCHAR(20) NOT NULL,
+    impact_score_llm FLOAT NOT NULL,
+    confidence_llm FLOAT NOT NULL,
+    confidence_calibrated FLOAT NOT NULL,
+    relevance_score FLOAT NOT NULL,
+    event_type VARCHAR(50) NOT NULL,
+    rule_sign INTEGER NOT NULL,
+    final_score FLOAT NOT NULL,
+
+    finbert_pos FLOAT NOT NULL,
+    finbert_neu FLOAT NOT NULL,
+    finbert_neg FLOAT NOT NULL,
+
+    reasoning_json TEXT,
+    model_fast VARCHAR(100),
+    model_reliable VARCHAR(100),
+    scored_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+
+    CONSTRAINT uq_news_sentiments_v2_processed_horizon UNIQUE (news_processed_id, horizon_days)
+);
+
+CREATE INDEX IF NOT EXISTS ix_news_sentiments_v2_processed_scored
+ON news_sentiments_v2 (news_processed_id, scored_at);
+
+CREATE INDEX IF NOT EXISTS ix_news_sentiments_v2_final_score
+ON news_sentiments_v2 (final_score);
+
+CREATE INDEX IF NOT EXISTS ix_news_sentiments_v2_label
+ON news_sentiments_v2 (label);
+
+CREATE TABLE IF NOT EXISTS daily_sentiments_v2 (
+    id BIGSERIAL PRIMARY KEY,
+    date TIMESTAMPTZ NOT NULL UNIQUE,
+    sentiment_index FLOAT NOT NULL,
+    news_count INTEGER NOT NULL DEFAULT 0,
+    avg_confidence FLOAT,
+    avg_relevance FLOAT,
+    source_version VARCHAR(20) NOT NULL DEFAULT 'v2',
+    aggregated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS ix_daily_sentiments_v2_date
+ON daily_sentiments_v2 (date);
+
+CREATE INDEX IF NOT EXISTS ix_daily_sentiments_v2_index
+ON daily_sentiments_v2 (sentiment_index);
+
+ALTER TABLE pipeline_run_metrics
+ADD COLUMN IF NOT EXISTS articles_scored_v2 INTEGER;
+
+ALTER TABLE pipeline_run_metrics
+ADD COLUMN IF NOT EXISTS llm_parse_fail_count INTEGER;
+
+ALTER TABLE pipeline_run_metrics
+ADD COLUMN IF NOT EXISTS escalation_count INTEGER;
+
+ALTER TABLE pipeline_run_metrics
+ADD COLUMN IF NOT EXISTS fallback_count INTEGER;
+
+COMMIT;
+

worker/tasks.py

@@ -386,16 +386,30 @@ async def _execute_pipeline_stages_v2(
         session.rollback()
     
     # -------------------------------------------------------------------------
-    # Stage 2: Sentiment scoring (existing - uses news_articles for now)
+    # Stage 2: Sentiment scoring (V2 - news_processed based)
     # -------------------------------------------------------------------------
     logger.info(f"[run_id={run_id}] Stage 2: Sentiment scoring")
     try:
-        from app.ai_engine import score_unscored_articles
-        
-        scored = score_unscored_articles(session)
+        from app.ai_engine import score_unscored_processed_articles
+
+        scoring_stats = score_unscored_processed_articles(session)
+        session.commit()
+
+        result["articles_scored"] = int(scoring_stats.get("scored_count", 0))
+        result["articles_scored_v2"] = int(scoring_stats.get("scored_count", 0))
+        result["llm_parse_fail_count"] = int(scoring_stats.get("parse_fail_count", 0))
+        result["escalation_count"] = int(scoring_stats.get("escalation_count", 0))
+        result["fallback_count"] = int(scoring_stats.get("fallback_count", 0))
+
+        update_run_metrics(
+            session,
+            run_id,
+            articles_scored_v2=result["articles_scored_v2"],
+            llm_parse_fail_count=result["llm_parse_fail_count"],
+            escalation_count=result["escalation_count"],
+            fallback_count=result["fallback_count"],
+        )
         session.commit()
-        
-        result["articles_scored"] = scored
         
     except Exception as e:
         logger.error(f"[run_id={run_id}] Stage 2 failed: {e}")
@@ -407,12 +421,14 @@ async def _execute_pipeline_stages_v2(
     # -------------------------------------------------------------------------
     logger.info(f"[run_id={run_id}] Stage 3: Sentiment aggregation")
     try:
-        from app.ai_engine import aggregate_daily_sentiment
-        
+        from app.ai_engine import aggregate_daily_sentiment, aggregate_daily_sentiment_v2
+
+        days_aggregated_v2 = aggregate_daily_sentiment_v2(session)
         days_aggregated = aggregate_daily_sentiment(session)
         session.commit()
         
         result["days_aggregated"] = days_aggregated
+        result["days_aggregated_v2"] = days_aggregated_v2
         
     except Exception as e:
         logger.error(f"[run_id={run_id}] Stage 3 failed: {e}")