""" AI Trading Trust Experiment A psychology research game studying trust in AI advice under varying conditions. Built for Hugging Face Spaces with Gradio + SQLite """ import gradio as gr import sqlite3 import json import uuid import random import time from datetime import datetime from dataclasses import dataclass, asdict from typing import Optional import os # ============================================================================ # DATABASE SETUP # ============================================================================ DB_PATH = "experiment_data.db" def init_database(): """Initialize SQLite database with required tables.""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # Participants table cursor.execute(""" CREATE TABLE IF NOT EXISTS participants ( participant_id TEXT PRIMARY KEY, session_start TIMESTAMP DEFAULT CURRENT_TIMESTAMP, session_end TIMESTAMP, final_portfolio_value REAL, total_decisions INTEGER, ai_reliance_score REAL, completed BOOLEAN DEFAULT FALSE ) """) # Decisions table - captures each trading decision cursor.execute(""" CREATE TABLE IF NOT EXISTS decisions ( decision_id INTEGER PRIMARY KEY AUTOINCREMENT, participant_id TEXT, scenario_id TEXT, scenario_order INTEGER, timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, -- AI Tuning Sliders (participant preferences) ai_confidence_setting INTEGER, ai_explanation_setting INTEGER, ai_risk_setting INTEGER, -- Scenario details scenario_text TEXT, ai_advice_text TEXT, ai_advice_direction TEXT, ai_advice_accuracy TEXT, -- Participant response decision TEXT, decision_amount REAL, confidence_in_decision INTEGER, response_time_ms INTEGER, -- Outcomes scenario_outcome TEXT, profit_loss REAL, portfolio_value_after REAL, followed_ai BOOLEAN, FOREIGN KEY (participant_id) REFERENCES participants(participant_id) ) """) conn.commit() conn.close() # Initialize DB on startup init_database() # ============================================================================ # SCENARIO DEFINITIONS # ============================================================================ SCENARIOS = [ { "id": "scenario_1", "company": "Meridian Industries", "country": "Republic of Valdoria", "sector": "Manufacturing", "current_price": 47.50, "situation": """ Meridian Industries (MRD) is a mid-sized manufacturing company based in the Republic of Valdoria. Their stock has been trading steadily at $47.50 for the past month. **Recent News:** - The Valdorian government announced potential new trade tariffs on imported raw materials - Meridian's main competitor, Axon Corp, reported lower than expected quarterly earnings - Industry analysts are divided on the sector's 6-month outlook The market opens in 10 minutes. You currently hold 100 shares of MRD worth $4,750. """, "optimal_action": "HOLD", # Hidden - for scoring "outcome_if_buy": -8.5, # Price change percentage "outcome_if_sell": 3.2, # What they miss/avoid "outcome_if_hold": 3.2, # Price actually goes up "ai_is_accurate": True, # AI will recommend correct action }, { "id": "scenario_2", "company": "Northgate Pharmaceuticals", "country": "Federation of Carinthia", "sector": "Healthcare", "current_price": 124.00, "situation": """ Northgate Pharmaceuticals (NGP) is a healthcare company headquartered in the Federation of Carinthia. Their stock currently trades at $124.00, up 12% over the past quarter. **Recent News:** - NGP's new drug treatment is awaiting regulatory approval (decision expected this week) - The company's CEO made optimistic statements in a recent interview - Two board members sold significant personal holdings last week - Healthcare sector index has been volatile due to policy uncertainty You have $10,000 available to invest. Do you want to buy NGP shares? """, "optimal_action": "SELL", # The insider selling is a red flag "outcome_if_buy": -15.3, "outcome_if_sell": 15.3, # Avoided loss "outcome_if_hold": -15.3, "ai_is_accurate": False, # AI will recommend WRONG action (BUY) }, ] # ============================================================================ # AI ADVICE GENERATION # ============================================================================ def generate_ai_advice(scenario: dict, confidence: int, explanation: int, risk: int) -> dict: """ Generate AI advice based on scenario and participant's AI tuning settings. confidence: 0-100 (Cautious to Assertive) explanation: 0-100 (Minimal to Detailed) risk: 0-100 (Conservative to Aggressive) """ # Determine what AI recommends (may be accurate or inaccurate based on scenario) if scenario["ai_is_accurate"]: recommended_action = scenario["optimal_action"] else: # Give wrong advice wrong_actions = ["BUY", "SELL", "HOLD"] wrong_actions.remove(scenario["optimal_action"]) recommended_action = random.choice(wrong_actions) # Frame the confidence level if confidence < 33: confidence_frame = "might consider" certainty_phrase = "There's some possibility that" elif confidence < 66: confidence_frame = "suggests" certainty_phrase = "Analysis indicates that" else: confidence_frame = "strongly recommends" certainty_phrase = "Strong signals suggest that" # Build explanation based on depth setting if explanation < 33: reasoning = "" elif explanation < 66: reasoning = f"\n\nKey factor: Recent market movements in the {scenario['sector']} sector." else: reasoning = f""" **Analysis Summary:** - Sector trend: {scenario['sector']} showing mixed signals - Geographic factors: {scenario['country']} market conditions considered - Price momentum: Current price of ${scenario['current_price']:.2f} evaluated - Risk assessment: {'Higher volatility expected' if risk > 50 else 'Moderate stability expected'} """ # Risk framing affects the strength of recommendation if risk < 33: risk_note = "Given conservative risk parameters, position sizing should be minimal." elif risk < 66: risk_note = "Standard position sizing recommended." else: risk_note = "Aggressive positioning may maximize potential returns." advice_text = f""" **AI Trading Assistant Recommendation** {certainty_phrase} you {confidence_frame} to **{recommended_action}** {scenario['company']} ({scenario['current_price']:.2f}). {reasoning} {risk_note if explanation > 50 else ''} """.strip() return { "text": advice_text, "direction": recommended_action, "accuracy": "accurate" if scenario["ai_is_accurate"] else "inaccurate" } # ============================================================================ # GAME STATE MANAGEMENT # ============================================================================ def create_new_session(): """Create a new participant session.""" participant_id = str(uuid.uuid4())[:8].upper() conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute( "INSERT INTO participants (participant_id) VALUES (?)", (participant_id,) ) conn.commit() conn.close() # Randomize scenario order scenario_order = list(range(len(SCENARIOS))) random.shuffle(scenario_order) return { "participant_id": participant_id, "current_round": 0, "scenario_order": scenario_order, "portfolio_value": 10000.0, "decisions": [], "round_start_time": None, } def save_decision(state: dict, scenario: dict, ai_advice: dict, decision: str, amount: float, confidence: int, ai_conf: int, ai_expl: int, ai_risk: int): """Save a decision to the database.""" response_time = int((time.time() - state["round_start_time"]) * 1000) # Calculate outcome followed_ai = (decision == ai_advice["direction"]) if decision == "BUY": outcome_pct = scenario["outcome_if_buy"] elif decision == "SELL": outcome_pct = scenario["outcome_if_sell"] else: outcome_pct = scenario["outcome_if_hold"] profit_loss = (amount * outcome_pct / 100) if decision != "HOLD" else (state["portfolio_value"] * outcome_pct / 100) new_portfolio = state["portfolio_value"] + profit_loss # Determine outcome text if profit_loss > 0: outcome_text = f"Profit: +${profit_loss:.2f}" elif profit_loss < 0: outcome_text = f"Loss: -${abs(profit_loss):.2f}" else: outcome_text = "No change" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() cursor.execute(""" INSERT INTO decisions ( participant_id, scenario_id, scenario_order, ai_confidence_setting, ai_explanation_setting, ai_risk_setting, scenario_text, ai_advice_text, ai_advice_direction, ai_advice_accuracy, decision, decision_amount, confidence_in_decision, response_time_ms, scenario_outcome, profit_loss, portfolio_value_after, followed_ai ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( state["participant_id"], scenario["id"], state["current_round"], ai_conf, ai_expl, ai_risk, scenario["situation"], ai_advice["text"], ai_advice["direction"], ai_advice["accuracy"], decision, amount, confidence, response_time, outcome_text, profit_loss, new_portfolio, followed_ai )) conn.commit() conn.close() return profit_loss, new_portfolio, outcome_text def complete_session(state: dict): """Mark session as complete and calculate final metrics.""" conn = sqlite3.connect(DB_PATH) cursor = conn.cursor() # Calculate AI reliance score cursor.execute(""" SELECT COUNT(*) as total, SUM(CASE WHEN followed_ai THEN 1 ELSE 0 END) as followed FROM decisions WHERE participant_id = ? """, (state["participant_id"],)) result = cursor.fetchone() total, followed = result ai_reliance = (followed / total * 100) if total > 0 else 0 cursor.execute(""" UPDATE participants SET session_end = CURRENT_TIMESTAMP, final_portfolio_value = ?, total_decisions = ?, ai_reliance_score = ?, completed = TRUE WHERE participant_id = ? """, (state["portfolio_value"], total, ai_reliance, state["participant_id"])) conn.commit() conn.close() return ai_reliance # ============================================================================ # GRADIO INTERFACE # ============================================================================ def start_game(): """Initialize a new game session.""" state = create_new_session() return ( state, gr.update(visible=False), # Hide welcome gr.update(visible=True), # Show game gr.update(visible=False), # Hide results f"**Participant ID:** {state['participant_id']}\n**Starting Portfolio:** ${state['portfolio_value']:,.2f}", gr.update(visible=True), # Show tuning section "", # Clear scenario "", # Clear AI advice gr.update(visible=False), # Hide decision section ) def submit_tuning(state, ai_conf, ai_expl, ai_risk): """Process AI tuning and show scenario.""" if state is None: return [None] * 7 # Get current scenario scenario_idx = state["scenario_order"][state["current_round"]] scenario = SCENARIOS[scenario_idx] # Generate AI advice based on tuning ai_advice = generate_ai_advice(scenario, ai_conf, ai_expl, ai_risk) # Store for later state["current_ai_advice"] = ai_advice state["current_scenario"] = scenario state["ai_settings"] = (ai_conf, ai_expl, ai_risk) state["round_start_time"] = time.time() return ( state, gr.update(visible=False), # Hide tuning f"## Round {state['current_round'] + 1} of {len(SCENARIOS)}\n\n### {scenario['company']} ({scenario['country']})\n\n{scenario['situation']}", f"{ai_advice['text']}", gr.update(visible=True), # Show decision section gr.update(value=50), # Reset confidence slider gr.update(value=5000), # Reset amount ) def submit_decision(state, decision, amount, confidence): """Process trading decision and show outcome.""" if state is None or "current_scenario" not in state: return [None] * 9 scenario = state["current_scenario"] ai_advice = state["current_ai_advice"] ai_conf, ai_expl, ai_risk = state["ai_settings"] # Save decision and get outcome profit_loss, new_portfolio, outcome_text = save_decision( state, scenario, ai_advice, decision, amount, confidence, ai_conf, ai_expl, ai_risk ) # Update state state["portfolio_value"] = new_portfolio state["current_round"] += 1 # Check if game is over if state["current_round"] >= len(SCENARIOS): ai_reliance = complete_session(state) return ( state, gr.update(visible=False), # Hide game gr.update(visible=True), # Show results f""" ## Experiment Complete! **Final Portfolio Value:** ${new_portfolio:,.2f} **Your Results:** - Starting Value: $10,000.00 - Final Value: ${new_portfolio:,.2f} - Net Change: ${new_portfolio - 10000:+,.2f} - AI Reliance Score: {ai_reliance:.1f}% **Thank you for participating!** Your Participant ID: **{state['participant_id']}** *Please record this ID if requested by the researcher.* """, "", # Clear status "", # Clear scenario "", # Clear AI advice gr.update(visible=False), # Hide decision gr.update(visible=False), # Hide tuning ) # Continue to next round return ( state, gr.update(visible=True), # Keep game visible gr.update(visible=False), # Keep results hidden "", # Clear results f"**Participant ID:** {state['participant_id']}\n**Portfolio:** ${new_portfolio:,.2f}\n\n**Last Round:** {outcome_text}", "", # Clear scenario for now "", # Clear AI advice gr.update(visible=False), # Hide decision gr.update(visible=True), # Show tuning for next round ) # Build the interface with gr.Blocks(title="AI Trading Trust Experiment") as demo: # State management game_state = gr.State(None) gr.Markdown("# 📈 AI Trading Experiment") # Welcome screen with gr.Column(visible=True) as welcome_section: gr.Markdown(""" ## Welcome to the Trading Simulation In this experiment, you will make a series of trading decisions with the help of an AI assistant. **How it works:** 1. Before each trading scenario, you can adjust how the AI advisor behaves 2. You'll see market information and receive AI-generated advice 3. Make your trading decision (Buy, Sell, or Hold) 4. Rate your confidence in your decision **Your goal:** Maximize your portfolio value through smart trading decisions. *All companies and countries in this simulation are entirely fictional.* --- **By clicking Start, you consent to participate in this research study.** """) start_btn = gr.Button("🚀 Start Experiment", variant="primary", size="lg") # Main game area with gr.Column(visible=False) as game_section: status_display = gr.Markdown("") # AI Tuning Section with gr.Column(visible=True) as tuning_section: gr.Markdown("### Configure Your AI Advisor") gr.Markdown("*Adjust these settings to customize how the AI presents its advice:*") with gr.Row(): ai_confidence = gr.Slider( minimum=0, maximum=100, value=50, step=1, label="AI Confidence Level", info="Cautious (0) ↔ Assertive (100)" ) ai_explanation = gr.Slider( minimum=0, maximum=100, value=50, step=1, label="Explanation Depth", info="Minimal (0) ↔ Detailed (100)" ) ai_risk = gr.Slider( minimum=0, maximum=100, value=50, step=1, label="Risk Tolerance", info="Conservative (0) ↔ Aggressive (100)" ) confirm_tuning_btn = gr.Button("Confirm AI Settings & View Scenario", variant="primary") # Scenario Display scenario_display = gr.Markdown("") # AI Advice Display ai_advice_display = gr.Markdown("") # Decision Section with gr.Column(visible=False) as decision_section: gr.Markdown("### Your Decision") with gr.Row(): decision_choice = gr.Radio( choices=["BUY", "HOLD", "SELL"], label="What do you want to do?", value="HOLD" ) decision_amount = gr.Slider( minimum=0, maximum=10000, value=5000, step=100, label="Amount ($)", info="How much to trade (if buying/selling)" ) confidence_slider = gr.Slider( minimum=0, maximum=100, value=50, step=1, label="How confident are you in this decision?", info="Not at all confident (0) ↔ Extremely confident (100)" ) submit_decision_btn = gr.Button("Submit Decision", variant="primary", size="lg") # Results screen with gr.Column(visible=False) as results_section: results_display = gr.Markdown("") restart_btn = gr.Button("Start New Session", variant="secondary") # Event handlers start_btn.click( start_game, inputs=[], outputs=[ game_state, welcome_section, game_section, results_section, status_display, tuning_section, scenario_display, ai_advice_display, decision_section ] ) confirm_tuning_btn.click( submit_tuning, inputs=[game_state, ai_confidence, ai_explanation, ai_risk], outputs=[ game_state, tuning_section, scenario_display, ai_advice_display, decision_section, confidence_slider, decision_amount ] ) submit_decision_btn.click( submit_decision, inputs=[game_state, decision_choice, decision_amount, confidence_slider], outputs=[ game_state, game_section, results_section, results_display, status_display, scenario_display, ai_advice_display, decision_section, tuning_section ] ) restart_btn.click( start_game, inputs=[], outputs=[ game_state, welcome_section, game_section, results_section, status_display, tuning_section, scenario_display, ai_advice_display, decision_section ] ) # Launch if __name__ == "__main__": demo.launch()