"""
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)
Reference in New Issue View Git Blame Copy Permalink