""" Tikker ML Service Microservice for machine learning-based keystroke analytics. Provides pattern detection, anomaly detection, and behavioral analysis. """ from fastapi import FastAPI, HTTPException, Query from pydantic import BaseModel from typing import Dict, List, Any, Optional import logging import os from ml_analytics import KeystrokeAnalyzer, MLPredictor, Pattern, Anomaly logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI( title="Tikker ML Service", description="Machine learning analytics for keystroke data", version="1.0.0" ) analyzer = KeystrokeAnalyzer( db_path=os.getenv("DB_PATH", "tikker.db") ) predictor = MLPredictor() class KeystrokeEvent(BaseModel): timestamp: int key_code: int event_type: str class PatternDetectionRequest(BaseModel): events: List[Dict[str, Any]] user_id: Optional[str] = "default" class AnomalyDetectionRequest(BaseModel): events: List[Dict[str, Any]] user_id: Optional[str] = "default" class BehavioralProfileRequest(BaseModel): events: List[Dict[str, Any]] user_id: Optional[str] = "default" class AuthenticityCheckRequest(BaseModel): events: List[Dict[str, Any]] user_id: Optional[str] = "default" class TemporalAnalysisRequest(BaseModel): date_range_days: int = 7 class PatternResponse(BaseModel): name: str confidence: float frequency: int description: str features: Dict[str, Any] class AnomalyResponse(BaseModel): timestamp: str anomaly_type: str severity: float reason: str expected_value: float actual_value: float class HealthResponse(BaseModel): status: str ml_available: bool api_version: str @app.get("/health", response_model=HealthResponse) async def health_check() -> HealthResponse: """Health check endpoint.""" return HealthResponse( status="healthy", ml_available=True, api_version="1.0.0" ) @app.post("/patterns/detect", response_model=List[PatternResponse]) async def detect_patterns(request: PatternDetectionRequest) -> List[PatternResponse]: """ Detect typing patterns in keystroke data. Identifies patterns such as: - Fast vs slow typing - Consistent vs inconsistent rhythm - Specialized typing behaviors """ try: if not request.events: raise HTTPException(status_code=400, detail="Events cannot be empty") patterns = analyzer.detect_patterns(request.events) return [ PatternResponse( name=p.name, confidence=p.confidence, frequency=p.frequency, description=p.description, features=p.features ) for p in patterns ] except HTTPException: raise except Exception as e: logger.error(f"Pattern detection error: {e}") raise HTTPException(status_code=500, detail=f"Pattern detection failed: {str(e)}") @app.post("/anomalies/detect", response_model=List[AnomalyResponse]) async def detect_anomalies(request: AnomalyDetectionRequest) -> List[AnomalyResponse]: """ Detect anomalies in keystroke behavior. Compares current behavior against baseline profile to identify: - Unusual typing speed - Abnormal rhythm patterns - Behavioral deviations """ try: if not request.events: raise HTTPException(status_code=400, detail="Events cannot be empty") anomalies = analyzer.detect_anomalies(request.events, request.user_id) return [ AnomalyResponse( timestamp=a.timestamp, anomaly_type=a.anomaly_type, severity=a.severity, reason=a.reason, expected_value=a.expected_value, actual_value=a.actual_value ) for a in anomalies ] except HTTPException: raise except Exception as e: logger.error(f"Anomaly detection error: {e}") raise HTTPException(status_code=500, detail=f"Anomaly detection failed: {str(e)}") @app.post("/profile/build") async def build_behavioral_profile(request: BehavioralProfileRequest) -> Dict[str, Any]: """ Build comprehensive behavioral profile from keystroke data. Creates a baseline profile containing: - Average typing speed - Peak activity hours - Common words - Consistency score - Detected patterns """ try: if not request.events: raise HTTPException(status_code=400, detail="Events cannot be empty") profile = analyzer.build_behavioral_profile(request.events, request.user_id) return { "user_id": profile.user_id, "avg_typing_speed": profile.avg_typing_speed, "peak_hours": profile.peak_hours, "common_words": profile.common_words, "consistency_score": profile.consistency_score, "patterns": profile.patterns } except HTTPException: raise except Exception as e: logger.error(f"Profile building error: {e}") raise HTTPException(status_code=500, detail=f"Profile building failed: {str(e)}") @app.post("/authenticity/check") async def check_authenticity(request: AuthenticityCheckRequest) -> Dict[str, Any]: """ Check if keystroke pattern matches known user profile. Returns authenticity score and verdict: - authentic: High confidence match - likely_authentic: Good confidence match - uncertain: Moderate confidence - suspicious: Low confidence match """ try: if not request.events: raise HTTPException(status_code=400, detail="Events cannot be empty") result = analyzer.predict_user_authenticity(request.events, request.user_id) return result except HTTPException: raise except Exception as e: logger.error(f"Authenticity check error: {e}") raise HTTPException(status_code=500, detail=f"Authenticity check failed: {str(e)}") @app.post("/temporal/analyze") async def analyze_temporal_patterns(request: TemporalAnalysisRequest) -> Dict[str, Any]: """ Analyze temporal patterns in keystroke data. Identifies trends over time: - Increasing/decreasing activity - Daily patterns - Weekly trends """ try: result = analyzer.analyze_temporal_patterns(request.date_range_days) return result except Exception as e: logger.error(f"Temporal analysis error: {e}") raise HTTPException(status_code=500, detail=f"Temporal analysis failed: {str(e)}") @app.post("/model/train") async def train_model( sample_size: int = Query(100, ge=10, le=10000) ) -> Dict[str, Any]: """ Train ML model on historical keystroke data. Parameters: - sample_size: Number of samples to use for training """ try: training_data = [{"typing_speed": 50 + i} for i in range(sample_size)] result = predictor.train_model(training_data) return result except Exception as e: logger.error(f"Model training error: {e}") raise HTTPException(status_code=500, detail=f"Model training failed: {str(e)}") @app.post("/behavior/predict") async def predict_behavior(request: PatternDetectionRequest) -> Dict[str, Any]: """ Predict user behavior based on trained ML model. Classifies behavior into categories: - normal: Expected behavior - fast_focused: Fast, focused typing - slow_deliberate: Careful, deliberate typing - stressed_or_tired: Inconsistent rhythm """ try: if not request.events: raise HTTPException(status_code=400, detail="Events cannot be empty") result = predictor.predict_behavior(request.events) return result except HTTPException: raise except Exception as e: logger.error(f"Behavior prediction error: {e}") raise HTTPException(status_code=500, detail=f"Behavior prediction failed: {str(e)}") @app.get("/") async def root() -> Dict[str, Any]: """Root endpoint with service information.""" return { "name": "Tikker ML Service", "version": "1.0.0", "status": "running", "ml_available": True, "endpoints": { "health": "/health", "patterns": "/patterns/detect", "anomalies": "/anomalies/detect", "profile": "/profile/build", "authenticity": "/authenticity/check", "temporal": "/temporal/analyze", "model": "/model/train", "behavior": "/behavior/predict" } } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8003)