safety_layer.py

"""
safety_layer.py — Response Safety Scanner
============================================
Scans all outgoing responses for risky financial claims
and automatically adds disclaimers / softens language.

Runs as the LAST step before returning response to frontend.
"""

import re
import logging
from typing import Dict, Any, List, Tuple

logger = logging.getLogger("TradeSense.Safety")

# ---------------------------------------------------------------------------
# RISKY PHRASES — patterns that indicate dangerous financial advice
# ---------------------------------------------------------------------------

RISKY_PHRASES: List[Tuple[str, str]] = [
    # (regex pattern, replacement or flag-only marker)
    (r"\bguaranteed?\s+(?:profit|return|income|gain)", "potential profit"),
    (r"\b100\s*%\s*(?:return|profit|sure|certain|guaranteed)", "strong potential return"),
    (r"\bsure\s*(?:shot|fire|thing|bet)\b", "promising opportunity"),
    (r"\brisk[\s-]*free\b", "lower-risk"),
    (r"\bcan'?t\s+(?:lose|fail|go\s+wrong)\b", "has historically performed well"),
    (r"\ball[\s-]*in\s+(?:invest|buy|put)\b", "consider a significant position"),
    (r"\bno\s+risk\b", "manageable risk"),
    (r"\bdouble\s+your\s+money\b", "grow your investment"),
    (r"\bget\s+rich\s+(?:quick|fast)\b", "build wealth over time"),
    (r"\bnever\s+(?:lose|losing|lost)\b", "historically resilient"),
    (r"\balways\s+(?:goes?\s+up|profit|win)\b", "has shown an upward trend"),
    (r"\bfool[\s-]*proof\b", "well-researched"),
    (r"\bprint\s+money\b", "generate returns"),
    (r"\bjackpot\b", "strong opportunity"),
]

# Standard disclaimer to append when risks detected
DISCLAIMER = (
    "\n\n⚠️ Disclaimer: Investing in financial markets involves risk. "
    "Past performance does not guarantee future results. "
    "Always do your own research (DYOR) and consider consulting a SEBI-registered "
    "financial advisor before making investment decisions."
)

# Shorter risk banner text for the meta field
RISK_BANNER_TEXT = "Investing involves risk. Always do your own research."


# ---------------------------------------------------------------------------
# CORE SCANNER
# ---------------------------------------------------------------------------

def scan_response(response: Dict[str, Any]) -> Dict[str, Any]:
    """
    Scan and sanitize a response dict before returning to frontend.
    
    Modifications:
    1. Scans 'insight' text for risky phrases and softens them.
    2. Adds disclaimer if risky content detected.
    3. Sets meta.has_risk_warning = True to trigger frontend banner.
    
    This function is IDEMPOTENT — safe to call multiple times.
    """
    if not isinstance(response, dict):
        return response
    
    insight = response.get("insight", "")
    has_risk = False
    
    if isinstance(insight, str) and insight.strip():
        modified_insight, risk_found = _sanitize_text(insight)
        
        if risk_found:
            has_risk = True
            # Append disclaimer if not already present
            if "⚠️ Disclaimer" not in modified_insight:
                modified_insight += DISCLAIMER
            response["insight"] = modified_insight
            logger.warning(f"Safety layer triggered: {risk_found} risky phrase(s) detected and softened.")
    
    # Also scan data-level text fields (e.g., education content)
    data = response.get("data", {})
    if isinstance(data, dict):
        for text_key in ["definition", "content", "description"]:
            text_val = data.get(text_key, "")
            if isinstance(text_val, str) and text_val.strip():
                modified_text, risk_found_data = _sanitize_text(text_val)
                if risk_found_data:
                    has_risk = True
                    data[text_key] = modified_text
    
    # Update meta with risk warning flag
    meta = response.get("meta", {})
    if not isinstance(meta, dict):
        meta = {}
    meta["has_risk_warning"] = has_risk
    response["meta"] = meta
    
    return response


def _sanitize_text(text: str) -> Tuple[str, int]:
    """
    Scan text for risky phrases and replace them.
    Returns (modified_text, count_of_replacements).
    """
    count = 0
    modified = text
    
    for pattern, replacement in RISKY_PHRASES:
        matches = re.findall(pattern, modified, re.IGNORECASE)
        if matches:
            count += len(matches)
            modified = re.sub(pattern, replacement, modified, flags=re.IGNORECASE)
    
    return modified, count


def add_investment_disclaimer(response: Dict[str, Any]) -> Dict[str, Any]:
    """
    Force-add a general investment disclaimer to strategy responses.
    Called specifically for strategy-type responses which always involve
    real money calculations.
    """
    insight = response.get("insight", "")
    if isinstance(insight, str) and "⚠️ Disclaimer" not in insight:
        response["insight"] = insight + (
            "\n\n⚠️ Note: This simulation uses historical data. "
            "Actual returns may vary due to brokerage fees, taxes, dividends, "
            "corporate actions, and market conditions."
        )
    
    meta = response.get("meta", {})
    if not isinstance(meta, dict):
        meta = {}
    meta["has_risk_warning"] = True
    response["meta"] = meta
    
    return response