Trading Education Platform - FIML-Powered Bot¶

# Educational lesson backed by real market data
from fiml.mcp import search_by_symbol
from fiml.agents import deep_equity_analysis

async def lesson_with_live_data(symbol: str):
    """Teach concepts using current market conditions"""

    # Get real-time snapshot via FIML
    data = await search_by_symbol(
        symbol=symbol,
        market="US",
        depth="standard"
    )

    # Generate educational narrative
    analysis = await deep_equity_analysis(
        symbol=symbol,
        include_narrative=True,
        include_recommendation=False  # Educational only
    )

    return {
        "current_price": data.price,
        "educational_context": analysis.narrative,
        "volatility_lesson": generate_volatility_lesson(data),
        "real_market_example": True
    }

from fiml.core.cache import TimeSeriesCache
from fiml.providers import get_historical_data

async def create_flash_crash_simulation():
    """Historical event replay for educational purposes"""

    # Fetch real historical data via FIML
    crash_data = await get_historical_data(
        symbol="SPY",
        start_date="2010-05-06",
        end_date="2010-05-07",
        interval="1m"
    )

    return {
        "event": "Flash Crash 2010",
        "real_data": True,
        "data_source": "FIML Multi-Provider",
        "educational_objectives": [
            "Understand circuit breakers",
            "Recognize liquidity crises",
            "Practice emotional discipline"
        ],
        "simulation_data": crash_data
    }

from fiml.mcp import MCPServer
from fiml.agents.narrative import NarrativeGenerator

class EducationalMentor:
    """AI mentor powered by FIML's narrative engine"""

    def __init__(self, persona: str):
        self.persona = persona  # Maya, Theo, or Zara
        self.narrative_gen = NarrativeGenerator()
        self.compliance_filter = ComplianceFilter()

    async def respond_to_question(self, question: str, context: dict):
        """Generate educational response with market context"""

        # Get relevant market data via FIML
        market_context = await self.get_market_context(context)

        # Generate narrative with compliance filtering
        response = await self.narrative_gen.generate(
            prompt=self.build_educational_prompt(question),
            market_data=market_context,
            persona=self.persona
        )

        # Apply safety filter
        safe_response = self.compliance_filter.apply(response)

        # Add mandatory disclaimer
        return f"{safe_response}\n\n⚠️ Educational purposes only — not investment advice."

from fiml.websocket import WebSocketStreamer

async def live_learning_session(symbols: list[str]):
    """Stream real market data for live lessons"""

    uri = "ws://fiml-server:8000/ws/prices/" + ",".join(symbols)

    async with websockets.connect(uri) as ws:
        print("🎓 Live Learning Session Started")

        while True:
            update = await ws.recv()
            data = json.loads(update)

            # Educational commentary on real price moves
            if abs(data['change_percent']) > 1.0:
                await send_educational_alert(
                    symbol=data['symbol'],
                    change=data['change_percent'],
                    lesson_context="volatility_basics"
                )

from fiml.features.session import SessionManager

class AdaptiveLearningEngine:
    """Track student progress using FIML's session system"""

    def __init__(self):
        self.session_mgr = SessionManager()

    async def track_lesson_completion(self, user_id: str, lesson_id: str):
        """Store progress and adapt difficulty"""

        session = await self.session_mgr.get_or_create(user_id)

        # Update mastery score
        session.data['mastery'][lesson_id] = self.calculate_mastery(
            recent_success=session.data['quiz_scores'][-5:],
            previous_mastery=session.data['mastery'].get(lesson_id, 0.5)
        )

        # Determine next lesson based on mastery
        next_lesson = self.select_next_lesson(
            mastery_score=session.data['mastery'][lesson_id],
            completed_lessons=session.data['completed'],
            error_patterns=session.data['errors']
        )

        await self.session_mgr.update(user_id, session.data)

        return next_lesson

def select_next_lesson(mastery: float, completed: set, errors: list):
    """FIML-powered adaptive path selection"""

    if mastery < 0.6:
        # Review mode - reinforce concepts
        return get_review_lesson(errors[-3:])

    elif 0.6 <= mastery < 0.8:
        # New content - appropriate difficulty
        return get_progressive_lesson(completed)

    else:
        # Mastery achieved - unlock advanced content
        return get_scenario_unlock(completed)

from fiml.features.compliance import ComplianceFramework

class EducationalComplianceLayer:
    """Ensure all content meets regulatory standards"""

    def __init__(self):
        self.compliance = ComplianceFramework()

    async def validate_educational_content(self, content: dict):
        """Screen content for compliance violations"""

        # Check for prohibited language
        violations = self.compliance.check_response(
            content['text'],
            user_region=content['user_region']
        )

        if violations:
            # Replace with safe educational content
            return self.generate_safe_alternative(content, violations)

        # Add region-appropriate disclaimer
        disclaimer = self.compliance.get_disclaimer(content['user_region'])
        content['text'] += f"\n\n{disclaimer}"

        return content

async def bitcoin_volatility_scenario(user_id: str):
    """Educational scenario using live BTC data"""

    # Fetch real BTC price via FIML
    btc_data = await search_by_coin(
        symbol="BTC",
        exchange="binance",
        pair="USDT"
    )

    scenario = {
        "title": "Bitcoin Weekend Volatility",
        "setup": f"""
            It's Friday evening. Bitcoin is trading at ${btc_data.price:,.2f}.
            You're monitoring the market when you notice increasing volatility.

            [Real current BTC data from FIML]

            What's your approach?
        """,

        "choices": [
            {
                "id": "watch_wait",
                "text": "Monitor the situation, take no action",
                "outcome": "cautious_observer",
                "teaches": "patience, avoiding FOMO"
            },
            {
                "id": "set_alerts",
                "text": "Set price alerts at ±5% levels",
                "outcome": "prepared_trader",
                "teaches": "risk management, automation"
            },
            {
                "id": "check_fundamentals",
                "text": "Research recent Bitcoin news and on-chain data",
                "outcome": "informed_decision",
                "teaches": "fundamental analysis, context"
            },
            {
                "id": "panic_response",
                "text": "Immediately react to price movements",
                "outcome": "emotional_trader",
                "teaches": "psychology, avoiding impulsive decisions"
            }
        ]
    }

    return scenario

class GamificationEngine:
    """Reward system integrated with FIML sessions"""

    XP_REWARDS = {
        "lesson_completed": 10,
        "quiz_perfect": 15,
        "simulation_completed": 25,
        "scenario_mastered": 30,
        "live_analysis_correct": 20,
        "streak_maintained": 5,
        "daily_quest": 50
    }

    async def award_xp(self, user_id: str, action: str, metadata: dict):
        """Award XP and update progression"""

        session = await session_manager.get(user_id)

        # Base XP
        xp = self.XP_REWARDS.get(action, 0)

        # Bonus multipliers
        if session.data['streak_days'] >= 7:
            xp *= 1.5  # Weekly streak bonus

        if metadata.get('difficulty') == 'advanced':
            xp *= 1.3  # Advanced content bonus

        # Update session
        session.data['total_xp'] += xp
        session.data['level'] = self.calculate_level(session.data['total_xp'])

        await session_manager.update(user_id, session.data)

        return {
            "xp_earned": xp,
            "total_xp": session.data['total_xp'],
            "new_level": session.data['level']
        }

class CostOptimizedDataAccess:
    """Minimize FIML API costs while maximizing learning value"""

    def __init__(self, user_tier: str):
        self.tier = user_tier

    async def get_educational_data(self, query: dict):
        """Tier-appropriate data access"""

        if self.tier == "free":
            # Use cached data, batch queries
            return await fiml.query(
                **query,
                cache_strategy="prefer_cache",
                max_age_minutes=15,
                providers=["yahoo_finance"]  # Free tier only
            )

        elif self.tier == "pro":
            # Real-time with cost controls
            return await fiml.query(
                **query,
                cache_strategy="fresh_when_available",
                max_age_minutes=1,
                providers=["yahoo_finance", "alpha_vantage"]
            )

        else:  # premium
            # Full FIML capabilities
            return await fiml.query(
                **query,
                cache_strategy="always_fresh",
                enable_arbitration=True,
                all_providers=True
            )

async def daily_learning_session(user_id: str):
    """FIML-powered daily quest"""

    session = await session_manager.get(user_id)

    # 1. Welcome with real market snapshot
    market_snapshot = await fiml.get_market_overview()

    greeting = f"""
    Good morning! 🌅

    Today's Market Snapshot (via FIML):
    • S&P 500: {market_snapshot['SPY']['change']:+.2f}%
    • Nasdaq: {market_snapshot['QQQ']['change']:+.2f}%
    • Bitcoin: ${market_snapshot['BTC']['price']:,.0f}

    Your Daily Quest:
    ✅ Complete 1 lesson (10 XP)
    ✅ Pass 1 quiz (15 XP)
    ✅ Practice with live data (20 XP)

    Current Streak: {session.data['streak_days']} days 🔥
    """

    # 2. Personalized lesson recommendation
    next_lesson = adaptive_engine.select_lesson(
        mastery_scores=session.data['mastery'],
        recent_errors=session.data['errors'][-5:],
        market_conditions=market_snapshot
    )

    # 3. Live practice opportunity
    if market_snapshot['volatility'] > 1.5:
        # High volatility - offer volatility lesson
        practice = generate_volatility_practice(market_snapshot)
    else:
        # Normal conditions - standard practice
        practice = generate_standard_practice(next_lesson)

    return {
        "greeting": greeting,
        "daily_lesson": next_lesson,
        "practice_mode": practice,
        "bonus_challenge": check_for_market_events(market_snapshot)
    }

async def live_practice_mode(user_id: str, symbols: list[str]):
    """Real-time market observation practice"""

    uri = f"ws://fiml-server:8000/ws/prices/{','.join(symbols)}"

    instructions = """
    🎓 Live Practice Mode

    Watch these stocks for the next 5 minutes. Your task:
    1. Identify which stock has the highest volatility
    2. Note any significant volume spikes
    3. Describe the overall trend direction

    [Real-time data streaming from FIML...]
    """

    async with websockets.connect(uri) as ws:
        start_time = time.time()
        observations = []

        while time.time() - start_time < 300:  # 5 minutes
            update = await ws.recv()
            data = json.loads(update)

            # Store for analysis
            observations.append(data)

            # Real-time educational hints
            if abs(data['change_percent']) > 0.5:
                hint = f"💡 {data['symbol']} just moved {data['change_percent']:+.2f}% - note this!"

        # Quiz after observation
        return generate_practice_quiz(observations)

class EducationalComplianceSystem:
    """Multi-layer safety using FIML compliance framework"""

    def __init__(self):
        self.fiml_compliance = ComplianceFramework()
        self.educational_rules = self.load_education_rules()

    async def validate_content(self, content: str, user_region: str):
        """Comprehensive compliance check"""

        # Layer 1: FIML base compliance
        fiml_check = await self.fiml_compliance.check_response(
            content,
            user_region=user_region
        )

        if not fiml_check.is_compliant:
            return self.generate_safe_alternative(content, fiml_check)

        # Layer 2: Educational-specific rules
        edu_check = self.check_educational_compliance(content)

        if not edu_check.is_safe:
            return self.rewrite_for_education(content, edu_check)

        # Layer 3: Add disclaimers
        return self.add_educational_disclaimers(content, user_region)

    def check_educational_compliance(self, content: str) -> ComplianceResult:
        """Educational-specific safety checks"""

        violations = []

        # Prohibited patterns
        if re.search(r'\b(buy|sell|invest in) [A-Z]{1,5}\b', content, re.I):
            violations.append("specific_recommendation")

        if re.search(r'(will|should) (rise|fall|moon|crash)', content, re.I):
            violations.append("price_prediction")

        if re.search(r'(guaranteed|certain|sure thing)', content, re.I):
            violations.append("false_certainty")

        if re.search(r'(leverage|margin|borrow)', content, re.I):
            violations.append("risky_instruments")

        return ComplianceResult(
            is_safe=len(violations) == 0,
            violations=violations
        )

SAFE_TEMPLATES = {
    "concept_explanation": """
        Here's the educational concept: {concept}

        Example: {example}

        Remember, this is for learning purposes. Always consult a financial 
        advisor before making investment decisions.
    """,

    "historical_context": """
        Historically, we observe that {observation}.

        In the case of {historical_example}, {what_happened}.

        This teaches us about {lesson}.

        ⚠️ Past performance doesn't guarantee future results.
    """,

    "scenario_outcome": """
        In this scenario, {action} led to {outcome}.

        The key lesson: {principle}

        In real trading, outcomes vary. This is educational only.
    """,

    "deflection": """
        I can't provide investment advice, but I can explain the concept:

        {educational_explanation}

        Would you like to learn more about {related_lesson}?
    """
}

async def handle_unsafe_query(user_id: str, query: str):
    """Detect and handle concerning user queries"""

    escalation_triggers = {
        "financial_distress": [
            "need money now",
            "pay rent",
            "in debt",
            "desperate"
        ],
        "leverage_questions": [
            "max leverage",
            "margin call",
            "borrow to trade"
        ],
        "specific_advice": [
            "what should I buy",
            "sell my",
            "which stock"
        ]
    }

    for trigger_type, patterns in escalation_triggers.items():
        if any(pattern in query.lower() for pattern in patterns):

            # Log for review
            await log_escalation(user_id, trigger_type, query)

            # Provide safe educational response
            return SAFE_RESPONSES[trigger_type]

    # Normal educational flow
    return await process_educational_query(query)

class PlatformAnalytics:
    """Track learning effectiveness and FIML usage"""

    async def track_session(self, user_id: str, session_data: dict):
        """Comprehensive session analytics"""

        metrics = {
            # Learning metrics
            "lessons_completed": len(session_data['completed']),
            "quiz_accuracy": calculate_accuracy(session_data['quiz_scores']),
            "time_per_lesson": session_data['avg_time'],
            "streak_days": session_data['streak_days'],

            # Engagement metrics
            "practice_mode_uses": session_data['practice_count'],
            "simulation_completions": session_data['simulations'],
            "mentor_interactions": session_data['ai_queries'],

            # FIML usage metrics
            "fiml_queries": session_data['fiml_queries'],
            "live_data_views": session_data['live_views'],
            "websocket_sessions": session_data['ws_sessions'],
            "data_freshness_avg": session_data['data_age_avg'],

            # Cost metrics
            "fiml_cache_hit_rate": session_data['cache_hits'] / session_data['total_queries'],
            "llm_token_usage": session_data['tokens_used'],
            "estimated_cost": self.calculate_cost(session_data)
        }

        return metrics

class CostOptimizer:
    """Minimize infrastructure costs while maximizing learning"""

    def __init__(self):
        self.fiml_cache_strategy = CacheStrategy()

    async def optimize_data_access(self, lesson_type: str, user_tier: str):
        """Smart data fetching based on context"""

        if lesson_type == "static_concept":
            # No need for real-time data
            return {
                "use_cached": True,
                "max_age_hours": 24,
                "provider": "mock"  # Use mock data for static lessons
            }

        elif lesson_type == "live_practice":
            # Real-time required
            return {
                "use_websocket": user_tier in ["pro", "premium"],
                "use_polling": user_tier == "free",
                "refresh_rate": 5000 if user_tier == "free" else 1000
            }

        elif lesson_type == "historical_simulation":
            # Historical data - cache aggressively
            return {
                "cache_key": f"sim_{lesson_type}_{lesson_id}",
                "ttl_days": 365,  # Historical data never changes
                "generate_once": True
            }

class ExperimentFramework:
    """Test educational approaches"""

    experiments = {
        "mentor_style": {
            "variants": ["formal", "casual", "technical"],
            "metric": "completion_rate",
            "duration_days": 14
        },

        "data_refresh_rate": {
            "variants": ["1min", "5min", "15min"],
            "metric": "engagement_score",
            "duration_days": 7
        },

        "gamification_intensity": {
            "variants": ["high", "medium", "low"],
            "metric": "retention_d7",
            "duration_days": 30
        }
    }