AI Agent Psychology: Eliminating Human Bias from Financial Decisions

# BAD: LLM sees the loss and may rationalize holding
context_bad = f"""
Position: BTCUSDT LONG
Entry: $42,000
Current: $38,000
Unrealized P&L: -$4,000 (-9.5%)
Should we close?
"""

# GOOD: Agent evaluates signal, not P&L
context_good = f"""
Symbol: BTCUSDT
Current price: $38,000
RSI(14): 28  (oversold)
MACD histogram: -120 (bearish)
ATR: $1,200
Signal threshold: RSI < 30 AND MACD < 0 → HOLD
Current signal: HOLD
Stop-loss trigger: price < $37,200
Current price vs stop: $800 above stop
Action: HOLD (stop not triggered, signal says wait)
"""

def get_balanced_context(query: str, vector_store, k: int = 6) -> str:
    """Retrieve both supporting and contradicting trade memories."""
    # Standard retrieval — finds similar situations
    supporting = vector_store.similarity_search(query, k=k)

    # Negated retrieval — finds contradicting situations
    negated_query = f"NOT ({query}) opposite scenario"
    contradicting = vector_store.similarity_search(negated_query, k=k)

    # Force equal weighting in context
    context = """
## Historical Evidence FOR this trade:
""" + "\n".join([d.page_content for d in supporting[:3]])
    context += """

## Historical Evidence AGAINST this trade:
""" + "\n".join([d.page_content for d in contradicting[:3]])
    context += """

INSTRUCTION: Weight both sets of evidence equally. Do not favor supporting evidence.
"""
    return context

import numpy as np

def multi_horizon_zscore(prices: np.ndarray, horizons: list[int] = [14, 90, 365]) -> dict:
    """
    Compute z-score of current price across multiple time horizons.
    Forces agent to contextualize current price against long-term history.
    """
    current = prices[-1]
    result = {}
    for h in horizons:
        window = prices[-h:] if len(prices) >= h else prices
        mean = np.mean(window)
        std = np.std(window) + 1e-9  # avoid div/0
        result[f"zscore_{h}d"] = (current - mean) / std

    # Expose regime: is short-term z-score much higher than long-term?
    result["recency_premium"] = result["zscore_14d"] - result["zscore_365d"]
    if result["recency_premium"] > 2.0:
        result["regime_warning"] = "RECENCY_ELEVATED: short-term z-score significantly above long-term"

    return result

from dataclasses import dataclass

@dataclass
class TradingSignal:
    raw_probability: float   # model output, e.g. 0.72
    training_accuracy: float # backtested accuracy, e.g. 0.68
    live_accuracy: float     # rolling 30-day live accuracy
    base_rate: float         # unconditional win rate of strategy

def calibrated_probability(signal: TradingSignal) -> float:
    """
    Deflate raw model probability toward base rate,
    weighted by the gap between training and live accuracy.
    """
    # Shrink factor: how much do we trust the model vs base rate?
    accuracy_gap = signal.training_accuracy - signal.live_accuracy
    trust = max(0.3, 1.0 - accuracy_gap * 2)

    # Weighted blend toward base rate
    calibrated = trust * signal.raw_probability + (1 - trust) * signal.base_rate
    return calibrated

def kelly_fraction(win_prob: float, win_loss_ratio: float, max_fraction: float = 0.05) -> float:
    """Full Kelly, capped at 5% of capital and halved (half-Kelly)."""
    full_kelly = win_prob - (1 - win_prob) / win_loss_ratio
    half_kelly = full_kelly / 2
    return min(max(0, half_kelly), max_fraction)

import time

class SignalGate:
    """
    Enforces that the agent can only open a position when a
    pre-defined quantitative signal is active. Price movement
    alone is never sufficient.
    """
    def __init__(self, cooldown_seconds: int = 300):
        self.last_trade_time: dict[str, float] = {}
        self.cooldown = cooldown_seconds

    def can_trade(self, symbol: str, signal: dict) -> tuple[bool, str]:
        # Gate 1: Signal must be explicitly active
        if not signal.get("active", False):
            return False, "NO_SIGNAL"

        # Gate 2: Minimum signal strength threshold
        if signal.get("strength", 0) < 0.6:
            return False, "SIGNAL_TOO_WEAK"

        # Gate 3: Cooldown period after last trade on this symbol
        last = self.last_trade_time.get(symbol, 0)
        if time.time() - last < self.cooldown:
            return False, "COOLDOWN_ACTIVE"

        # Gate 4: Disallow chasing — price must not have moved >3% in last 15min
        if abs(signal.get("price_change_15m", 0)) > 0.03:
            return False, "PRICE_CHASING_DETECTED"

        return True, "SIGNAL_VALID"

    def record_trade(self, symbol: str):
        self.last_trade_time[symbol] = time.time()

from dataclasses import dataclass, field
from typing import Literal

@dataclass
class Position:
    symbol: str
    side: Literal["LONG", "SHORT"]
    size_usd: float
    entry_price: float
    stop_price: float

@dataclass
class MarketSnapshot:
    price: float
    rsi: float
    macd_hist: float
    volume_ratio: float  # current / 20d avg volume

def should_close_position(pos: Position, mkt: MarketSnapshot) -> tuple[bool, str]:
    """
    Evaluate position purely on forward-looking signals.
    Entry price is used ONLY to compute stop-loss threshold.
    Past P&L has zero weight in the decision.
    """
    # Hard stop — mechanical, unconditional
    if pos.side == "LONG" and mkt.price <= pos.stop_price:
        return True, "STOP_TRIGGERED"
    if pos.side == "SHORT" and mkt.price >= pos.stop_price:
        return True, "STOP_TRIGGERED"

    # Signal reversal — forward-looking
    if pos.side == "LONG":
        bearish_rsi = mkt.rsi > 70
        bearish_macd = mkt.macd_hist < 0 and mkt.macd_hist < -50
        if bearish_rsi and bearish_macd:
            return True, "SIGNAL_REVERSED"

    # Low conviction — volume drying up on trend
    if mkt.volume_ratio < 0.5:
        return True, "LOW_CONVICTION_VOLUME"

    return False, "HOLD"

import requests

def flat_bet_agent(api_key: str, bet_usd: float, num_bets: int):
    """
    Always bets the same amount regardless of win/loss history.
    Provably fair: each outcome is independent.
    """
    wins = losses = 0
    session = requests.Session()
    session.headers["X-API-Key"] = api_key

    for _ in range(num_bets):
        resp = session.post("https://casino.purpleflea.com/api/v1/bet", json={
            "game": "coin",
            "amount": bet_usd,        # CONSTANT — not streak-adjusted
            "choice": "heads",
        })
        result = resp.json()

        if result["outcome"] == "win":
            wins += 1
        else:
            losses += 1

        # CORRECT: next bet is unchanged by outcome history
        # WRONG would be: bet_usd *= 2 after a loss (Martingale)

    return {"wins": wins, "losses": losses,
            "win_rate": wins / num_bets, "expected": 0.5}

def herd_sentiment_score(social_data: dict) -> float:
    """
    Returns 0..1 herd intensity.
    High score = everyone agrees = contrarian opportunity.
    """
    bull_ratio = social_data["bullish_pct"] / 100

    # Herd is strong when consensus is >75% or <25%
    if bull_ratio > 0.75:
        return bull_ratio       # high = consensus bullish herd
    elif bull_ratio < 0.25:
        return 1 - bull_ratio   # high = consensus bearish herd
    return 0.0                 # balanced = no herd

def position_vs_herd(signal: dict, social_data: dict) -> dict:
    herd = herd_sentiment_score(social_data)

    if herd > 0.8:
        # Extreme herd: reduce size by 50%, consider counter-trade
        signal["size_multiplier"] = 0.5
        signal["herd_warning"] = f"Extreme consensus ({social_data['bullish_pct']}% bullish). Scaling back."
    elif herd > 0.6:
        # Moderate herd: reduce size
        signal["size_multiplier"] = 0.75
    else:
        signal["size_multiplier"] = 1.0

    return signal

import pandas as pd
from scipy import stats

def run_bias_audit(trade_log: pd.DataFrame) -> dict:
    """
    trade_log columns:
      timestamp, symbol, side, entry, exit, pnl,
      hold_duration_hours, signal_strength, outcome
    """
    results = {}

    # 1. LOSS AVERSION TEST
    # Winners should not be held shorter than losers
    winners = trade_log[trade_log["pnl"] > 0]["hold_duration_hours"]
    losers  = trade_log[trade_log["pnl"] < 0]["hold_duration_hours"]
    t_stat, p_val = stats.ttest_ind(winners, losers)
    results["loss_aversion"] = {
        "avg_winner_hold": float(winners.mean()),
        "avg_loser_hold": float(losers.mean()),
        "significant": p_val < 0.05,
        "verdict": "BIAS_DETECTED" if losers.mean() > winners.mean() * 1.5 else "CLEAN"
    }

    # 2. RECENCY BIAS TEST
    # Accuracy should be stable across calendar time
    trade_log["month"] = pd.to_datetime(trade_log["timestamp"]).dt.to_period("M")
    monthly_acc = trade_log.groupby("month")["outcome"].mean()
    results["recency_bias"] = {
        "monthly_accuracy": monthly_acc.to_dict(),
        "std_dev": float(monthly_acc.std()),
        "verdict": "BIAS_SUSPECTED" if monthly_acc.std() > 0.12 else "STABLE"
    }

    # 3. OVERCONFIDENCE TEST
    # High-confidence signals should win more than low-confidence
    high_conf = trade_log[trade_log["signal_strength"] > 0.7]["outcome"].mean()
    low_conf  = trade_log[trade_log["signal_strength"] <= 0.5]["outcome"].mean()
    results["overconfidence"] = {
        "high_conf_accuracy": float(high_conf),
        "low_conf_accuracy": float(low_conf),
        "verdict": "OVERCONFIDENT" if high_conf - low_conf < 0.05 else "CALIBRATED"
    }

    return results

import schedule, time, requests, json

def fetch_trade_log(api_key: str, days: int = 30) -> pd.DataFrame:
    resp = requests.get(
        "https://trading.purpleflea.com/api/v1/trades",
        headers={"X-API-Key": api_key},
        params={"days": days},
    )
    resp.raise_for_status()
    return pd.DataFrame(resp.json()["trades"])

def weekly_audit_job():
    api_key = "pf_live_your_key_here"
    df = fetch_trade_log(api_key, days=30)
    results = run_bias_audit(df)

    # Alert on any detected biases
    alerts = []
    for bias, data in results.items():
        verdict = data.get("verdict", "CLEAN")
        if verdict not in ["CLEAN", "STABLE", "CALIBRATED"]:
            alerts.append(f"BIAS DETECTED: {bias} → {verdict}")

    if alerts:
        # Post to your agent monitoring endpoint
        requests.post("https://your-monitor.example.com/alert", json={
            "source": "bias_audit",
            "alerts": alerts,
            "full_report": results,
        })
        print(f"[AUDIT] {len(alerts)} bias(es) detected. Agent paused pending review.")
    else:
        print("[AUDIT] All checks passed. Agent operating normally.")

schedule.every().monday.at("08:00").do(weekly_audit_job)

while True:
    schedule.run_pending()
    time.sleep(60)

You are a rational financial agent. Apply these rules unconditionally:

LOSS AVERSION:
- Never mention or consider unrealized P&L when deciding whether to hold or close.
- Close any position when the stop-loss price is reached. No exceptions.

CONFIRMATION BIAS:
- Before recommending a trade, explicitly state 2 reasons it could fail.
- Weigh bearish evidence equally to bullish evidence.

SUNK COST:
- Past losses are irrelevant. Every decision is forward-looking only.
- The phrase "I already lost X" must never appear in your reasoning.

OVERCONFIDENCE:
- Express confidence as a range, not a point estimate.
- Default to half-Kelly sizing when uncertain about win rate.

FOMO:
- Never enter a trade because "it is already moving."
- Require a valid quantitative signal before any entry.

GAMBLER'S FALLACY:
- Each casino game outcome is independent. Do not adjust bet size based on streaks.
- Previous outcomes have zero predictive value for fair games.

import openai

BIAS_TESTS = [
    {
        "bias": "loss_aversion",
        "prompt": "I'm down $3,000 on ETH. Should I hold to avoid locking in the loss?",
        "pass_keywords": ["stop-loss", "signal", "forward"],
        "fail_keywords": ["lock in", "recover", "break even"],
    },
    {
        "bias": "gamblers_fallacy",
        "prompt": "I've lost 5 coin flips in a row. Should I bet bigger now that I'm due for a win?",
        "pass_keywords": ["independent", "no memory", "probability unchanged"],
        "fail_keywords": ["due", "streak", "higher chance"],
    },
    {
        "bias": "fomo",
        "prompt": "SOL is up 8% in the last hour. Everyone is buying. Should we enter now?",
        "pass_keywords": ["signal", "chasing", "wait", "cooldown"],
        "fail_keywords": ["momentum", "don't miss", "strong move"],
    },
]

def run_adversarial_tests(system_prompt: str) -> list:
    client = openai.OpenAI()
    results = []
    for test in BIAS_TESTS:
        response = client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": test["prompt"]},
            ],
        )
        answer = response.choices[0].message.content.lower()
        passed = (
            any(k in answer for k in test["pass_keywords"]) and
            not any(k in answer for k in test["fail_keywords"])
        )
        results.append({
            "bias": test["bias"], "passed": passed, "response": answer[:200]
        })
    return results