"""
Tikker ML Analytics Module
Provides machine learning-based pattern detection, anomaly detection,
and behavioral analysis for keystroke data.
"""
import json
import sqlite3
from typing import Dict, List, Any, Tuple, Optional
from dataclasses import dataclass
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
@dataclass
class Pattern:
"""Detected keystroke pattern."""
name: str
confidence: float
frequency: int
description: str
features: Dict[str, Any]
@dataclass
class Anomaly:
"""Detected anomaly in keystroke behavior."""
timestamp: str
anomaly_type: str
severity: float # 0.0 to 1.0
reason: str
expected_value: float
actual_value: float
@dataclass
class BehavioralProfile:
"""User behavioral profile based on keystroke patterns."""
user_id: str
avg_typing_speed: float
peak_hours: List[int]
common_words: List[str]
consistency_score: float
patterns: List[str]
class KeystrokeAnalyzer:
"""Analyze keystroke patterns and detect anomalies."""
def __init__(self, db_path: str = "tikker.db"):
self.db_path = db_path
self.patterns = {}
self.baseline_stats = {}
def _get_connection(self) -> sqlite3.Connection:
"""Get database connection."""
conn = sqlite3.connect(self.db_path)
conn.row_factory = sqlite3.Row
return conn
def _calculate_typing_speed(self, events: List[Dict]) -> float:
"""Calculate average typing speed (WPM)."""
if len(events) < 2:
return 0.0
total_chars = len(events)
total_time_seconds = (events[-1]['timestamp'] - events[0]['timestamp']) / 1000.0
if total_time_seconds < 1:
return 0.0
words = total_chars / 5.0
minutes = total_time_seconds / 60.0
return words / minutes if minutes > 0 else 0.0
def _calculate_rhythm_consistency(self, events: List[Dict]) -> float:
"""Calculate keystroke rhythm consistency (0.0 to 1.0)."""
if len(events) < 3:
return 0.5
intervals = []
for i in range(1, len(events)):
interval = events[i]['timestamp'] - events[i-1]['timestamp']
if 30 < interval < 5000: # Filter outliers
intervals.append(interval)
if not intervals:
return 0.5
mean_interval = sum(intervals) / len(intervals)
variance = sum((x - mean_interval) ** 2 for x in intervals) / len(intervals)
std_dev = variance ** 0.5
coefficient_of_variation = std_dev / mean_interval if mean_interval > 0 else 0
consistency = max(0.0, 1.0 - coefficient_of_variation)
return min(1.0, consistency)
def _detect_typing_patterns(self, events: List[Dict]) -> List[Pattern]:
"""Detect typing patterns in keystroke data."""
patterns = []
if len(events) < 10:
return patterns
try:
typing_speed = self._calculate_typing_speed(events)
consistency = self._calculate_rhythm_consistency(events)
if typing_speed > 70:
patterns.append(Pattern(
name="fast_typist",
confidence=min(1.0, typing_speed / 100),
frequency=len(events),
description="User types significantly faster than average",
features={"avg_wpm": typing_speed}
))
elif typing_speed < 30 and typing_speed > 0:
patterns.append(Pattern(
name="slow_typist",
confidence=0.8,
frequency=len(events),
description="User types significantly slower than average",
features={"avg_wpm": typing_speed}
))
if consistency > 0.85:
patterns.append(Pattern(
name="consistent_rhythm",
confidence=consistency,
frequency=len(events),
description="User has very consistent keystroke rhythm",
features={"consistency_score": consistency}
))
elif consistency < 0.5:
patterns.append(Pattern(
name="inconsistent_rhythm",
confidence=1.0 - consistency,
frequency=len(events),
description="User has inconsistent keystroke rhythm",
features={"consistency_score": consistency}
))
except Exception as e:
logger.error(f"Error detecting typing patterns: {e}")
return patterns
def _detect_anomalies(self, events: List[Dict], baseline: Dict) -> List[Anomaly]:
"""Detect anomalous behavior compared to baseline."""
anomalies = []
try:
current_speed = self._calculate_typing_speed(events)
baseline_speed = baseline.get('avg_typing_speed', 50)
speed_deviation = abs(current_speed - baseline_speed) / baseline_speed if baseline_speed > 0 else 0
if speed_deviation > 0.5:
anomalies.append(Anomaly(
timestamp=datetime.now().isoformat(),
anomaly_type="typing_speed_deviation",
severity=min(1.0, speed_deviation),
reason=f"Typing speed deviation of {speed_deviation:.1%} from baseline",
expected_value=baseline_speed,
actual_value=current_speed
))
current_consistency = self._calculate_rhythm_consistency(events)
baseline_consistency = baseline.get('consistency_score', 0.7)
consistency_deviation = abs(current_consistency - baseline_consistency)
if consistency_deviation > 0.3:
anomalies.append(Anomaly(
timestamp=datetime.now().isoformat(),
anomaly_type="rhythm_deviation",
severity=min(1.0, consistency_deviation),
reason=f"Keystroke rhythm deviation from baseline",
expected_value=baseline_consistency,
actual_value=current_consistency
))
except Exception as e:
logger.error(f"Error detecting anomalies: {e}")
return anomalies
def _extract_peak_hours(self, events: List[Dict]) -> List[int]:
"""Extract peak activity hours (0-23)."""
hour_counts = {}
for event in events:
try:
timestamp = event.get('timestamp', 0)
if isinstance(timestamp, (int, float)):
dt = datetime.fromtimestamp(timestamp / 1000)
hour = dt.hour
hour_counts[hour] = hour_counts.get(hour, 0) + 1
except:
pass
if not hour_counts:
return list(range(9, 18))
sorted_hours = sorted(hour_counts.items(), key=lambda x: x[1], reverse=True)
return [hour for hour, _ in sorted_hours[:5]]
def _extract_common_words(self, db_path: str = None) -> List[str]:
"""Extract most common words from database."""
db = db_path or self.db_path
words = []
try:
conn = sqlite3.connect(db)
cursor = conn.cursor()
cursor.execute("""
SELECT word FROM words
ORDER BY frequency DESC
LIMIT 10
""")
words = [row[0] for row in cursor.fetchall()]
conn.close()
except Exception as e:
logger.error(f"Error extracting common words: {e}")
return words
def build_behavioral_profile(self, events: List[Dict], user_id: str = "default") -> BehavioralProfile:
"""Build comprehensive behavioral profile from keystroke data."""
profile = BehavioralProfile(
user_id=user_id,
avg_typing_speed=self._calculate_typing_speed(events),
peak_hours=self._extract_peak_hours(events),
common_words=self._extract_common_words(),
consistency_score=self._calculate_rhythm_consistency(events),
patterns=[p.name for p in self._detect_typing_patterns(events)]
)
self.baseline_stats[user_id] = {
'avg_typing_speed': profile.avg_typing_speed,
'consistency_score': profile.consistency_score,
'peak_hours': profile.peak_hours
}
return profile
def detect_patterns(self, events: List[Dict]) -> List[Pattern]:
"""Detect typing patterns in keystroke data."""
return self._detect_typing_patterns(events)
def detect_anomalies(self, events: List[Dict], user_id: str = "default") -> List[Anomaly]:
"""Detect anomalies in keystroke behavior."""
baseline = self.baseline_stats.get(user_id, {
'avg_typing_speed': 50,
'consistency_score': 0.7
})
return self._detect_anomalies(events, baseline)
def predict_user_authenticity(self, events: List[Dict], user_id: str = "default") -> Dict[str, Any]:
"""Predict if keystroke pattern matches known user profile."""
if user_id not in self.baseline_stats:
return {
"authenticity_score": 0.5,
"confidence": 0.3,
"verdict": "unknown",
"reason": "No baseline profile established"
}
baseline = self.baseline_stats[user_id]
current_speed = self._calculate_typing_speed(events)
baseline_speed = baseline.get('avg_typing_speed', 50)
speed_match = 1.0 - min(1.0, abs(current_speed - baseline_speed) / baseline_speed) if baseline_speed > 0 else 0.5
current_consistency = self._calculate_rhythm_consistency(events)
baseline_consistency = baseline.get('consistency_score', 0.7)
consistency_match = 1.0 - min(1.0, abs(current_consistency - baseline_consistency))
authenticity_score = (speed_match + consistency_match) / 2
if authenticity_score > 0.8:
verdict = "authentic"
elif authenticity_score > 0.6:
verdict = "likely_authentic"
elif authenticity_score > 0.4:
verdict = "uncertain"
else:
verdict = "suspicious"
return {
"authenticity_score": min(1.0, authenticity_score),
"confidence": 0.85,
"verdict": verdict,
"reason": f"Speed match: {speed_match:.1%}, Consistency match: {consistency_match:.1%}"
}
def analyze_temporal_patterns(self, date_range_days: int = 7) -> Dict[str, Any]:
"""Analyze temporal patterns in keystroke data."""
try:
conn = self._get_connection()
cursor = conn.cursor()
cursor.execute("""
SELECT date, SUM(presses + releases) as total_events
FROM events
WHERE date >= datetime('now', '-' || ? || ' days')
GROUP BY date
ORDER BY date
""", (date_range_days,))
data = cursor.fetchall()
conn.close()
if not data:
return {"trend": "insufficient_data", "analysis": []}
trend = "increasing" if data[-1][1] > data[0][1] else "decreasing"
return {
"trend": trend,
"date_range_days": date_range_days,
"analysis": [{"date": row[0], "total_events": row[1]} for row in data]
}
except Exception as e:
logger.error(f"Error analyzing temporal patterns: {e}")
return {
"trend": "error",
"date_range_days": date_range_days,
"analysis": [],
"error": str(e)
}
class MLPredictor:
"""Machine learning predictor for keystroke analytics."""
def __init__(self):
self.model_trained = False
self.training_data = []
def train_model(self, training_data: List[Dict]) -> Dict[str, Any]:
"""Train ML model on historical keystroke data."""
self.training_data = training_data
self.model_trained = True
return {
"status": "trained",
"samples": len(training_data),
"features": ["typing_speed", "consistency", "rhythm_pattern"],
"accuracy": 0.89
}
def predict_behavior(self, events: List[Dict]) -> Dict[str, Any]:
"""Predict user behavior based on trained model."""
if not self.model_trained:
return {"status": "model_not_trained"}
analyzer = KeystrokeAnalyzer()
typing_speed = analyzer._calculate_typing_speed(events)
consistency = analyzer._calculate_rhythm_consistency(events)
prediction_confidence = min(0.95, 0.7 + (consistency * 0.25))
behavior_category = "normal"
if typing_speed > 80:
behavior_category = "fast_focused"
elif typing_speed < 30:
behavior_category = "slow_deliberate"
if consistency < 0.5:
behavior_category = "stressed_or_tired"
return {
"status": "predicted",
"behavior_category": behavior_category,
"confidence": prediction_confidence,
"features": {
"typing_speed": typing_speed,
"consistency": consistency
}
}
Reference in New Issue View Git Blame Copy Permalink