Secure RAG for Trading and Fraud Detection with VectaX

pip install mirror-sdk mirror_enc
pip install mirror-sdk[examples]  # for OpenAI and ChromaDB examples

MIRROR_API_KEY=your-api-key
MIRROR_SERVER_URL=https://mirrorapi.azure-api.net/v1
MIRROR_TELEMETRY_ENABLED=true
MIRROR_POLICY_EVAL_ENABLED=true

from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig

config = MirrorConfig.from_env()
sdk = MirrorSDK(config)
print("Mirror SDK ready for financial services pipeline")

import openai
from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig
from mirror_sdk.core.models import VectorData

config = MirrorConfig.from_env()
sdk = MirrorSDK(config)

# A representative analyst trade note (reference ID only, no customer name in the vector ID)
trade_note = """
Trade: Long 50,000 shares NVDA @ 142.30. Entry thesis: data center buildout
accelerating, supply constraints easing Q3. Stop: 138.00. Target: 162.00.
Risk/reward 3.9:1. Position size 2.1% of book. Correlated with AMD long.
"""

# Step 1: Embed the trade note
response = openai.embeddings.create(
    model="text-embedding-3-small",
    input=trade_note
)
embedding = response.data[0].embedding

# Step 2: Encrypt before storage
# Vector ID is a non-identifying reference, not the trader's name
vector = VectorData(vector=embedding, id="trade_eq_2026_0410_001")
encrypted_vector = sdk.vectax.encrypt(vector)

# Step 3: Set access policy scoped to the equities group
# Only equity_analyst role in equities group can decrypt
equities_policy = {
    "roles": ["equity_analyst", "portfolio_manager"],
    "groups": ["equities"],
    "departments": ["trading"]
}
sdk.set_policy(equities_policy)

print("Trade note encrypted and scoped to equities group")
print("Investment banking group cannot decrypt this vector")

from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig

config = MirrorConfig.from_env()
sdk = MirrorSDK(config)

# Customer record metadata - contains regulated identifiers
customer_record = {
    "source": "retail_brokerage",
    "account_number": "4920183756",
    "ssn_last4": "6842",
    "kyc_tier": "standard",
    "open_date": "2019-03-14"
}

# Generate FPE key and tweak, then encrypt
fpe_key = sdk.metadata.generate_key()
fpe_tweak = sdk.metadata.generate_tweak_from_data(customer_record)
encrypted_record = sdk.metadata.encrypt(customer_record, fpe_key, fpe_tweak)

print(f"Original account:   {customer_record['account_number']}")
print(f"Encrypted account:  {encrypted_record['account_number']}")
print(f"Format preserved:   still 10 digits, usable by downstream systems")

# Store fpe_key in your key vault (Azure Key Vault, AWS KMS, etc.)
# Never store it alongside the data

import openai
from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig
from mirror_sdk.core.models import VectorData

config = MirrorConfig.from_env()
sdk = MirrorSDK(config)

# Step 1: Store a known fraud pattern (done at pattern library build time)
fraud_pattern = """
Pattern: rapid small-value transactions under reporting threshold across
multiple accounts within 24 hours, followed by a single large consolidating
transfer. Accounts opened within 60 days. No prior transaction history.
"""

pattern_embedding = openai.embeddings.create(
    model="text-embedding-3-small",
    input=fraud_pattern
).data[0].embedding

# Encrypt and store the fraud pattern
# Scoped to fraud_analyst and compliance roles only
encrypted_pattern = sdk.vectax.encrypt(
    VectorData(vector=pattern_embedding, id="fraud_pattern_smurfing_v2")
)

# Step 2: At detection time, embed and encrypt the incoming transaction
incoming_tx = """
Customer account opened 22 days ago. 14 deposits ranging $890-$970
across 6 branch locations over 48 hours. Single outbound wire $12,400.
"""

tx_embedding = openai.embeddings.create(
    model="text-embedding-3-small",
    input=incoming_tx
).data[0].embedding

encrypted_tx = sdk.vectax.encrypt(
    VectorData(vector=tx_embedding, id="tx_check_live")
)

# Step 3: Similarity search runs on ciphertext
# The encrypted_tx is compared against encrypted fraud patterns
# The matching score is computed without decrypting either vector
# (pattern: encrypted_results = fraud_vector_db.query_encrypted(encrypted_tx, n_results=5))

print("Transaction embedded and encrypted for pattern matching")
print("Fraud pattern library stays encrypted throughout")
print("Similarity ranking identical to plaintext comparison")

from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig
from mirror_sdk.core.mirror_api_models import Action
import logging

logger = logging.getLogger("financial_ai_guard")
config = MirrorConfig.from_env()
sdk = MirrorSDK(config)

def validate_financial_response(
    analyst_query: str,
    retrieved_context: str,
    model_response: str
) -> dict:
    """
    Validate an AI-generated financial response.
    Checks for: hallucination, PII leakage, bias.
    Returns a validation report before the response reaches the user.
    """

    # 1. Check for hallucination against the retrieved context
    hallucination_result = sdk.agentiq.analyze_hallucination(
        input=analyst_query,
        output=model_response,
        context=retrieved_context,
        threshold=0.75  # stricter threshold for financial advice
    )

    is_hallucinated = False
    if hallucination_result.pairs:
        is_hallucinated = any(
            str(p.is_hallucination).lower() == 'true'
            for p in hallucination_result.pairs
        )

    # 2. Scan for PII leakage in the response
    # Customer data from retrieved context must not leak to other users
    pii_result = sdk.agentiq.detect_pii(
        text=model_response,
        pii_entities=["NAME", "ACCOUNT_NUMBER", "SSN", "EMAIL", "PHONE"],
        action=Action.REDACT
    )

    # 3. Log and return validation outcome
    if is_hallucinated:
        logger.warning("Hallucination detected in financial AI response. Blocking.")

    if pii_result.entities:
        logger.warning(
            f"PII detected in response: {[e.label for e in pii_result.entities]}"
        )

    return {
        "approved": not is_hallucinated and len(pii_result.entities) == 0,
        "hallucination_detected": is_hallucinated,
        "pii_detected": len(pii_result.entities) > 0,
        "safe_response": pii_result.redacted_text,
        "risk_score": pii_result.risk_score
    }

import openai
from mirror_sdk.core.mirror_core import MirrorSDK, MirrorConfig
from mirror_sdk.core.models import VectorData
from mirror_sdk.core.mirror_api_models import Action

config = MirrorConfig.from_env()
sdk = MirrorSDK(config)

def trading_rag_query(
    analyst_query: str,
    analyst_role: str,
    analyst_group: str,
    analyst_department: str,
    vector_store,
    n_results: int = 4
) -> dict:
    """
    Run a trading desk RAG query with encrypted retrieval.
    The analyst's group scope determines which records are decryptable.
    Responses are validated for hallucination and PII before return.
    """

    # 1. Generate a role-scoped key for this analyst
    analyst_key = sdk.rbac.generate_user_secret_key({
        "roles": [analyst_role],
        "groups": [analyst_group],
        "departments": [analyst_department]
    })

    # 2. Embed and encrypt the query
    query_emb = openai.embeddings.create(
        model="text-embedding-3-small",
        input=analyst_query
    ).data[0].embedding

    encrypted_q = sdk.vectax.encrypt(
        VectorData(vector=query_emb, id="query")
    )

    # 3. Search encrypted vector store
    # Records outside the analyst's key scope cannot be decrypted
    encrypted_results = vector_store.query_encrypted(
        encrypted_q, n_results=n_results
    )

    # 4. Decrypt results using the scoped key
    contexts = []
    for enc_result in encrypted_results:
        try:
            decrypted = sdk.vectax.decrypt(enc_result)
            contexts.append(decrypted.metadata.get("summary", ""))
        except Exception:
            # Record from a different group: cannot decrypt, skip silently
            pass

    context_block = "\n\n".join(contexts)

    # 5. Generate AI response
    prompt = f"""You are a trading desk research assistant.
Answer using only the trade notes and research provided. Do not add
information not present in the notes. If the notes do not address the
question, say so clearly.

Notes:
{context_block}

Query: {analyst_query}

Response:"""

    response = openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.1
    )
    model_response = response.choices[0].message.content

    # 6. Validate before returning
    validation = validate_financial_response(
        analyst_query, context_block, model_response
    )

    if not validation["approved"]:
        return {
            "response": "Response flagged for review. Please contact your compliance team.",
            "approved": False,
            "reason": validation
        }

    return {
        "response": validation["safe_response"],
        "approved": True
    }