app.py

import os
import time
import gradio as gr
import textwrap
from dataclasses import dataclass, field
from typing import Dict, Tuple

# ============================================================
# PFI Elite Access Control (per-user access codes)
# ============================================================

# Set in Hugging Face Space -> Settings -> Variables and secrets
# Example:
#   PFI_ACCESS_CODES = "PFI-EDIN-001,PFI-CLIENT-002,PFI-CLIENT-003"
#   PFI_DAILY_QUOTA = "3"
#   PFI_BRAND_CONTACT = "pfi@bpmred.academy"
#   PFI_MIN_CHARS = "40"

ACCESS_CODES_RAW = os.getenv("PFI_ACCESS_CODES", "").strip()
DAILY_QUOTA = int(os.getenv("PFI_DAILY_QUOTA", "3").strip() or "3")
BRAND_CONTACT = os.getenv("PFI_BRAND_CONTACT", "pfi@bpmred.academy").strip()
MIN_CHARS = int(os.getenv("PFI_MIN_CHARS", "40").strip() or "40")

# Normalized set of valid codes
VALID_CODES = {c.strip() for c in ACCESS_CODES_RAW.split(",") if c.strip()}

# 24h window (seconds)
WINDOW_SECONDS = 24 * 60 * 60


@dataclass
class UsageEntry:
    # timestamps of successful requests
    ts: list = field(default_factory=list)


@dataclass
class AccessState:
    # code -> usage
    usage: Dict[str, UsageEntry] = field(default_factory=dict)
    # tiny audit log (session-local)
    audit: list = field(default_factory=list)


def _now() -> float:
    return time.time()


def _clean_old(ts_list: list, now: float) -> list:
    """Keep only timestamps inside rolling 24h window."""
    cutoff = now - WINDOW_SECONDS
    return [t for t in ts_list if t >= cutoff]


def validate_code(code: str) -> Tuple[bool, str]:
    code = (code or "").strip()
    if not code:
        return False, "Access Code is required."
    if not VALID_CODES:
        # If admin forgot to set PFI_ACCESS_CODES, fail closed (safest)
        return False, "PFI is not configured for access codes yet. Please contact support."
    if code not in VALID_CODES:
        return False, "Invalid Access Code."
    return True, "Access granted."


# ============================================================
# Core PFI reasoning stub (preview-only, non-executive)
# Replace this later with your real model/API call.
# ============================================================

def pfi_reasoning_preview(question: str) -> str:
    q = (question or "").strip()
    if len(q) < MIN_CHARS:
        return textwrap.dedent(
            f"""
            [Input rejected]

            PFI requires a precise, high-density financial question (min {MIN_CHARS} characters).
            Ambiguous or underspecified inputs reduce analytical value.

            Use this template:
            - Objective:
            - Horizon:
            - Constraints (max drawdown / liquidity / taxes / jurisdiction):
            - Current exposures (concentration risk):
            - What "irreversible downside" means to you:
            """
        ).strip()

    response = f"""
    PFI STRUCTURAL ANALYSIS (PREVIEW · NON-EXECUTABLE)

    Question (received):
    - {q}

    Structural Map:
    1) Decision domain: capital allocation / risk architecture
    2) Time structure: near-term liquidity vs long-horizon convexity
    3) Constraint set: drawdown tolerance, cash-flow needs, optionality preservation
    4) Failure modes: forced selling, duration mismatch, correlation spikes, policy shock
    5) Control levers: rebalancing rules, hedges, reserve sizing, exposure caps

    Cognitive Decomposition:
    - Primary variables:
      • income stability / career risk
      • liquidity needs and timing
      • inflation/regime risk
      • concentration risk (single asset / geography / employer)
      • tax / jurisdiction constraints
    - Secondary dependencies:
      • correlation behavior under stress
      • funding liquidity vs market liquidity
      • refinancing risk / rate sensitivity
    - Irreversible downside candidates:
      • ruin risk (capital impairment that changes future opportunity set)
      • forced liquidation triggers
      • leverage or short-vol exposure under volatility expansion

    Next Questions (to deepen analysis):
    - Define max acceptable drawdown and time-to-recover.
    - Specify liquidity schedule (must-pay obligations by month/quarter).
    - List top 3 concentrated exposures and whether they can be reduced.
    - Clarify jurisdiction + tax constraints (capital gains, withholding).
    """

    return textwrap.dedent(response).strip()


# ============================================================
# Gated handler (enforces per-code quota)
# ============================================================

def gated_request(access_code: str, question: str, state: AccessState) -> Tuple[str, AccessState]:
    state = state or AccessState()
    now = _now()

    ok, msg = validate_code(access_code)
    code = (access_code or "").strip()

    # audit
    state.audit.append((int(now), "validate", code, ok))

    if not ok:
        locked = f"""
🔒 **PFI Licensed Access Required**

Reason: **{msg}**

To request a licensed Access Code, contact: **{BRAND_CONTACT}**
"""
        return textwrap.dedent(locked).strip(), state

    # init usage
    if code not in state.usage:
        state.usage[code] = UsageEntry(ts=[])

    # rolling window cleanup
    state.usage[code].ts = _clean_old(state.usage[code].ts, now)
    used = len(state.usage[code].ts)

    if used >= DAILY_QUOTA:
        remaining_time = int((state.usage[code].ts[0] + WINDOW_SECONDS) - now)
        hours = max(0, remaining_time // 3600)
        minutes = max(0, (remaining_time % 3600) // 60)

        quota_msg = f"""
⛔ **Daily quota reached** for this Access Code.

- Quota: **{DAILY_QUOTA} requests / 24h**
- Next reset in: **~{hours}h {minutes}m**

If you need expanded access, contact: **{BRAND_CONTACT}**
"""
        state.audit.append((int(now), "quota_block", code, used))
        return textwrap.dedent(quota_msg).strip(), state

    # Consume 1 quota
    state.usage[code].ts.append(now)
    state.audit.append((int(now), "quota_consume", code, used + 1))

    # Run preview reasoning
    out = pfi_reasoning_preview(question)

    # Add header with usage
    used_after = len(state.usage[code].ts)
    header = f"✅ Access Code: {code}  |  Usage: {used_after}/{DAILY_QUOTA} (rolling 24h)\n\n"
    return header + out, state


# ============================================================
# UI
# ============================================================

with gr.Blocks(theme=gr.themes.Soft()) as demo:
    gr.Markdown(
        """
# 🧠 Personal Financial Intelligence (PFI)
**High-density financial reasoning · Research Preview**

⚠️ *PFI is a licensed research interface.  
It does NOT provide financial advice, trading signals, or execute decisions.*
"""
    )

    # Session state (in-memory per user session)
    state = gr.State(AccessState())

    with gr.Row():
        access_code = gr.Textbox(
            label="PFI Access Code (Licensed)",
            placeholder="e.g., PFI-CLIENT-002",
            type="password",
        )

    with gr.Row():
        question_input = gr.Textbox(
            label="Your Question",
            placeholder=(
                "Write ONE precise, high-impact financial question.\n"
                "Include horizon, constraints, and context. Ambiguity reduces output quality."
            ),
            lines=4,
        )

    btn = gr.Button("Request Structural Analysis (Preview)")

    output_box = gr.Textbox(
        label="PFI Output (Exploratory · Non-Executable)",
        lines=10,
    )

    btn.click(
        fn=gated_request,
        inputs=[access_code, question_input, state],
        outputs=[output_box, state],
    )

    gr.Markdown(
        f"""
---
🔒 **Deeper Structural Reasoning — Licensed Layer**

Personalized constraints, scenario stress tests, and capital structure reasoning  
are available only under **PFI Licensed Access**.

👉 Request access: **{BRAND_CONTACT}**

---
### Disclaimer
PFI outputs are exploratory and informational only.  
No financial advice, trading signals, or decision execution is provided.

© 2026 BPM RED Academy · All rights reserved.
"""
    )

if __name__ == "__main__":
    demo.launch()