apps/quant-engine/app.py

"""
MARKETSCOPE — Quant Engine (Python)

Runs conformal prediction / quantile forests for scenario band generation.
All outputs are SCENARIO BANDS, not point forecasts.

Training_Gate: Human_Review_Required
Audit_Spec: 4b565498-9afc-4782-af4a-c6b11a5d0058
"""

import hashlib
import json
from datetime import datetime, timezone

import numpy as np
import requests
from flask import Flask, jsonify, request

app = Flask(__name__)

# ── Configuration ──────────────────────────────────────────────────────────────

WORM_ENDPOINT = "http://localhost:8090"
LOGIC_ENGINE_URL = "http://localhost:8080"

# ── Scenario Band Generator ────────────────────────────────────────────────────

def generate_scenario_bands(data: dict) -> dict:
    """
    Generate scenario bands from market data.
    Returns probabilistic scenarios, NOT point forecasts.
    """
    symbol = data.get("symbol", "UNKNOWN")
    values = data.get("values", [])

    if not values:
        return {
            "symbol": symbol,
            "error": "No data provided",
            "scenarios": []
        }

    arr = np.array(values)
    mean = float(np.mean(arr))
    std = float(np.std(arr))

    # Generate scenario bands (NOT predictions)
    scenarios = [
        {
            "scenario": "bull_case",
            "label": "Bull Case (75th percentile)",
            "range": [mean + std, mean + 2 * std],
            "probability": "low",
            "note": "Scenario band — NOT a prediction"
        },
        {
            "scenario": "base_case",
            "label": "Base Case (median)",
            "range": [mean - std * 0.5, mean + std * 0.5],
            "probability": "medium",
            "note": "Scenario band — NOT a prediction"
        },
        {
            "scenario": "bear_case",
            "label": "Bear Case (25th percentile)",
            "range": [mean - 2 * std, mean - std],
            "probability": "low",
            "note": "Scenario band — NOT a prediction"
        },
        {
            "scenario": "tail_risk",
            "label": "Tail Risk (5th percentile)",
            "range": [mean - 3 * std, mean - 2 * std],
            "probability": "very_low",
            "note": "Scenario band — NOT a prediction"
        }
    ]

    # Compute data hash for WORM logging
    data_hash = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()

    return {
        "symbol": symbol,
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "data_hash": data_hash,
        "statistics": {
            "mean": mean,
            "std": std,
            "min": float(np.min(arr)),
            "max": float(np.max(arr)),
            "count": len(values)
        },
        "scenarios": scenarios,
        "disclaimer": "SCENARIO BANDS — NOT PREDICTIONS — Human Review Required",
        "training_gate": "Human_Review_Required",
        "audit_spec": "4b565498-9afc-4782-af4a-c6b11a5d0058"
    }


def log_to_worm(result: dict) -> bool:
    """Log scenario bands to WORM chain."""
    try:
        response = requests.post(
            f"{WORM_ENDPOINT}/append_block",
            json={
                "block_type": "scenario_output",
                "symbol": result.get("symbol"),
                "data_hash": result.get("data_hash"),
                "timestamp": result.get("timestamp"),
                "scenario_count": len(result.get("scenarios", [])),
            },
            timeout=5
        )
        return response.status().is_success()
    except Exception:
        return False


# ── Routes ─────────────────────────────────────────────────────────────────────

@app.route("/health")
def health():
    return jsonify({
        "status": "healthy",
        "service": "marketscope-quant-engine",
        "version": "0.1.0",
        "output_type": "scenario_bands",
        "training_gate": "Human_Review_Required",
    })


@app.route("/scenarios", methods=["POST"])
def generate_scenarios():
    """
    Generate scenario bands from market data.
    Input: {"symbol": "SPX", "values": [100, 101, 99, ...]}
    Output: Scenario bands with disclaimer
    """
    data = request.get_json()

    if not data or "values" not in data:
        return jsonify({"error": "Missing 'values' in request body"}), 400

    result = generate_scenario_bands(data)

    # Log to WORM chain
    log_to_worm(result)

    return jsonify(result)


@app.route("/regime/<symbol>")
def get_regime(symbol: str):
    """
    Query Prolog Logic Engine for regime classification.
    Returns regime as scenario context, NOT as prediction.
    """
    try:
        response = requests.post(
            f"{LOGIC_ENGINE_URL}/query",
            json={
                "predicate": "detect_regime",
                "args": [symbol]
            },
            timeout=5
        )
        regime_data = response.json()

        return jsonify({
            "symbol": symbol,
            "regime": regime_data,
            "context": "Regime classification for scenario band generation",
            "disclaimer": "NOT a prediction — scenario context only",
        })
    except Exception as e:
        return jsonify({
            "symbol": symbol,
            "regime": "unknown",
            "error": str(e),
        })


@app.route("/compliance/check", methods=["POST"])
def check_compliance():
    """
    Check signal compliance against sovereign constraints.
    """
    signal = request.get_json()

    if not signal:
        return jsonify({"error": "Missing signal in request body"}), 400

    # Check for prohibited actions
    prohibited_types = ["buy_signal", "sell_signal", "price_target", "trade_recommendation"]
    signal_type = signal.get("type", "")

    if signal_type in prohibited_types:
        return jsonify({
            "compliant": False,
            "violation": "financial_advice_prohibited",
            "message": "This signal type is PROHIBITED under sovereign axioms",
            "audit_spec": "4b565498-9afc-4782-af4a-c6b11a5d0058",
        })

    return jsonify({
        "compliant": True,
        "message": "Signal passed compliance check",
        "training_gate": "Human_Review_Required",
    })


# ── Main ───────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8081, debug=False)