mirror of
https://github.com/velocitatem/PHANTOM.git
synced 2026-07-16 01:53:37 +00:00
Compare commits
1 Commits
34-initial
...
claude/hum
| Author | SHA1 | Date | |
|---|---|---|---|
|
aab54ea7c0 |
10
README.md
10
README.md
@@ -1,12 +1,8 @@
|
|||||||
<img width="200" align="left" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
|
|
||||||
|
|
||||||
### PHANTOM
|
<img width="1952" height="2176" alt="nobody_knows" src="https://github.com/user-attachments/assets/d148b00d-e9f9-4280-89cc-0cc866e17251" />
|
||||||
|
|
||||||
[](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
|
[](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
|
||||||
[](https://sites.research.google/trc/faq/)
|
|
||||||
[](https://phantom-hotel.vercel.app)
|
|
||||||
[](https://phantom-airline.vercel.app)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
- https://phantom-hotel.vercel.app/
|
||||||
|
- https://phantom-airline.vercel.app/
|
||||||
|
|
||||||
|
|||||||
@@ -1,15 +1,4 @@
|
|||||||
services:
|
services:
|
||||||
|
|
||||||
tensorboard:
|
|
||||||
image: tensorflow/tensorflow:latest
|
|
||||||
container_name: "PHANTOM-tensorboard"
|
|
||||||
ports:
|
|
||||||
- "6006:6006"
|
|
||||||
volumes:
|
|
||||||
- ./experiments/ml/runs:/logs
|
|
||||||
command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
|
|
||||||
restart: unless-stopped
|
|
||||||
|
|
||||||
backend:
|
backend:
|
||||||
container_name: "PHANTOM-backend"
|
container_name: "PHANTOM-backend"
|
||||||
build:
|
build:
|
||||||
@@ -134,7 +123,6 @@ services:
|
|||||||
- AIRFLOW__CORE__LOAD_EXAMPLES=false
|
- AIRFLOW__CORE__LOAD_EXAMPLES=false
|
||||||
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
|
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
|
||||||
- AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
|
- AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
|
||||||
- AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
|
|
||||||
- KAFKA_HOST=kafka
|
- KAFKA_HOST=kafka
|
||||||
- KAFKA_PORT=29092
|
- KAFKA_PORT=29092
|
||||||
- BACKEND_URL=http://backend:5000
|
- BACKEND_URL=http://backend:5000
|
||||||
@@ -170,7 +158,6 @@ services:
|
|||||||
- AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
|
- AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
|
||||||
- AIRFLOW__CORE__LOAD_EXAMPLES=false
|
- AIRFLOW__CORE__LOAD_EXAMPLES=false
|
||||||
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
|
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
|
||||||
- AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
|
|
||||||
- KAFKA_HOST=kafka
|
- KAFKA_HOST=kafka
|
||||||
- KAFKA_PORT=29092
|
- KAFKA_PORT=29092
|
||||||
- BACKEND_URL=http://backend:5000
|
- BACKEND_URL=http://backend:5000
|
||||||
|
|||||||
403
docs/ARCHITECTURE_OVERVIEW.md
Normal file
403
docs/ARCHITECTURE_OVERVIEW.md
Normal file
@@ -0,0 +1,403 @@
|
|||||||
|
# Multi-Task Learning Architecture - Quick Reference
|
||||||
|
|
||||||
|
## Current System (Baseline)
|
||||||
|
|
||||||
|
```
|
||||||
|
┌─────────────────────────────────────────────────────────────────┐
|
||||||
|
│ CURRENT STATE │
|
||||||
|
├─────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ Browser Events → Next.js → FastAPI → Kafka (user-interactions) │
|
||||||
|
│ ↓ │
|
||||||
|
│ Airflow (every 15min) │
|
||||||
|
│ ↓ │
|
||||||
|
│ [Messy SessionState Pipeline] │
|
||||||
|
│ ↓ │
|
||||||
|
│ Simple Rule-Based Pricing: │
|
||||||
|
│ - Surge (if demand > 10) │
|
||||||
|
│ - Elasticity formula │
|
||||||
|
│ - Velocity threshold for agents │
|
||||||
|
│ ↓ │
|
||||||
|
│ Redis (prices) │
|
||||||
|
│ ↓ │
|
||||||
|
│ Pricing Provider API │
|
||||||
|
│ │
|
||||||
|
│ ISSUES: │
|
||||||
|
│ ✗ O(n²) feature extraction │
|
||||||
|
│ ✗ No supervised ML for agent detection │
|
||||||
|
│ ✗ Simple heuristics (velocity > 5 → agent) │
|
||||||
|
│ ✗ No learning from data │
|
||||||
|
│ ✗ Margin leakage not effectively addressed │
|
||||||
|
└─────────────────────────────────────────────────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
## Proposed System (Multi-Task Learning)
|
||||||
|
|
||||||
|
```
|
||||||
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ PHASE 1: DATA PIPELINE │
|
||||||
|
├──────────────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ Kafka (user-interactions) │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌─────────────────────────────────────┐ │
|
||||||
|
│ │ VECTORIZED FEATURE PIPELINE │ │
|
||||||
|
│ ├─────────────────────────────────────┤ │
|
||||||
|
│ │ 1. TemporalFeatureExtractor │ → 8 features (velocity, etc.) │
|
||||||
|
│ │ 2. BehavioralFeatureExtractor │ → 10 features (carts, hovers) │
|
||||||
|
│ │ 3. ProductFeatureExtractor │ → 8 features (prices, depth) │
|
||||||
|
│ │ 4. UserAgentParser │ → 3 features (browser type) │
|
||||||
|
│ │ 5. SessionAggregator │ → Session-level matrix │
|
||||||
|
│ │ 6. ExperimentLabelJoiner │ → Join with xp_human_only │
|
||||||
|
│ └─────────────────────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ Feature Matrix: [sessionId, 29 features, 3 labels] │
|
||||||
|
│ │
|
||||||
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
|
|
||||||
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ PHASE 2: SUPERVISED AGENT CLASSIFIER │
|
||||||
|
├──────────────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ Feature Matrix (29 features) │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌────────────────────┐ │
|
||||||
|
│ │ XGBoost Model │ │
|
||||||
|
│ ├────────────────────┤ │
|
||||||
|
│ │ Input: 29 dims │ │
|
||||||
|
│ │ Output: P(agent) │ │
|
||||||
|
│ │ Loss: BCE │ │
|
||||||
|
│ └────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ Target: ROC-AUC > 0.90 │
|
||||||
|
│ │
|
||||||
|
│ DEPLOYMENT: │
|
||||||
|
│ - Real-time inference in Pricing Provider │
|
||||||
|
│ - Dynamic markup: P(agent) > 0.7 → 1.3x price │
|
||||||
|
│ - Retrain daily via Airflow │
|
||||||
|
│ │
|
||||||
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
|
|
||||||
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ PHASE 3: MULTI-TASK LEARNING MODEL │
|
||||||
|
├──────────────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ Input: Session Features (29) + Product Features (10) + Current Price │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌───────────────────────────────────────────────────────────┐ │
|
||||||
|
│ │ MULTI-TASK NEURAL NETWORK │ │
|
||||||
|
│ ├───────────────────────────────────────────────────────────┤ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ ┌──────────────────────┐ │ │
|
||||||
|
│ │ │ Session Encoder │ (Shared) │ │
|
||||||
|
│ │ │ [29] → [128] → [64] │ │ │
|
||||||
|
│ │ └──────────┬───────────┘ │ │
|
||||||
|
│ │ │ │ │
|
||||||
|
│ │ ├────────────┬───────────────┐ │ │
|
||||||
|
│ │ ↓ ↓ ↓ │ │
|
||||||
|
│ │ ┌─────────┐ ┌─────────┐ ┌─────────────┐ │ │
|
||||||
|
│ │ │ Task A │ │ Product │ │ Task B │ │ │
|
||||||
|
│ │ │ Agent │ │ Encoder │ │ Purchase │ │ │
|
||||||
|
│ │ │ Head │ │ [10]→16 │ │ Prob Head │ │ │
|
||||||
|
│ │ └────┬────┘ └────┬────┘ └──────┬──────┘ │ │
|
||||||
|
│ │ ↓ └────┬────────────┘ │ │
|
||||||
|
│ │ P(agent) ↓ │ │
|
||||||
|
│ │ P(purchase|price) │ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ Loss = α·BCE(agent) + β·BCE(purchase) │ │
|
||||||
|
│ │ α=1.0, β=2.0 (tune these weights) │ │
|
||||||
|
│ └───────────────────────────────────────────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ OUTPUTS: │
|
||||||
|
│ 1. Agent probability (like Phase 2) │
|
||||||
|
│ 2. Purchase probability given price │
|
||||||
|
│ 3. Session embedding (for knowledge distillation) │
|
||||||
|
│ │
|
||||||
|
│ USE CASE: │
|
||||||
|
│ Optimal Price = argmax_p [ p · P(purchase|p) · (1 + λ·P(agent)) ] │
|
||||||
|
│ │
|
||||||
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
|
|
||||||
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ KNOWLEDGE DISTILLATION BRANCH │
|
||||||
|
├──────────────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ Multi-Task Model (teacher) │
|
||||||
|
│ ↓ │
|
||||||
|
│ Generate predictions on validation set │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌──────────────────────────────────────┐ │
|
||||||
|
│ │ Distill to Decision Tree (student) │ │
|
||||||
|
│ ├──────────────────────────────────────┤ │
|
||||||
|
│ │ Input: 29 session features │ │
|
||||||
|
│ │ Output: Optimal markup multiplier │ │
|
||||||
|
│ │ Max depth: 5 (interpretable) │ │
|
||||||
|
│ └──────────────────────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ Extract Human-Readable Rules: │
|
||||||
|
│ │
|
||||||
|
│ IF interaction_velocity > 10 AND cart_to_view_ratio < 0.1: │
|
||||||
|
│ markup = 1.3 (likely agent reconnaissance) │
|
||||||
|
│ ELIF unique_products_viewed < 3 AND session_duration > 300: │
|
||||||
|
│ markup = 0.9 (engaged human, offer discount) │
|
||||||
|
│ ELSE: │
|
||||||
|
│ markup = 1.0 (baseline) │
|
||||||
|
│ │
|
||||||
|
│ Also: SHAP values for feature importance analysis │
|
||||||
|
│ │
|
||||||
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
|
|
||||||
|
┌──────────────────────────────────────────────────────────────────────────┐
|
||||||
|
│ PHASE 4: SYNTHETIC DYNAMIC PRICING SIMULATOR │
|
||||||
|
├──────────────────────────────────────────────────────────────────────────┤
|
||||||
|
│ │
|
||||||
|
│ PURPOSE: Fast experimentation without real users │
|
||||||
|
│ │
|
||||||
|
│ ┌────────────────────────────────────────────────────┐ │
|
||||||
|
│ │ DynamicPricingEnv (Gymnasium) │ │
|
||||||
|
│ ├────────────────────────────────────────────────────┤ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ State: [demand, inventory, hour, agent_frac, │ │
|
||||||
|
│ │ avg_velocity] │ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ Action: price_multiplier ∈ [0.7, 1.5] │ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ Dynamics: │ │
|
||||||
|
│ │ - Simulate user arrivals (Poisson) │ │
|
||||||
|
│ │ - Split into humans (30%) vs agents (70%) │ │
|
||||||
|
│ │ - Purchase probability: │ │
|
||||||
|
│ │ P_human(buy) = logistic(price, sensitivity=2) │ │
|
||||||
|
│ │ P_agent(buy) = logistic(price, sensitivity=5) │ │
|
||||||
|
│ │ │ │
|
||||||
|
│ │ Reward: revenue - 0.5 * margin_leakage │ │
|
||||||
|
│ │ where margin_leakage = (oracle_price - │ │
|
||||||
|
│ │ actual_price) × │ │
|
||||||
|
│ │ agent_purchases │ │
|
||||||
|
│ └────────────────────────────────────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ ┌────────────────────────────────────────┐ │
|
||||||
|
│ │ Train RL Agent (PPO) │ │
|
||||||
|
│ ├────────────────────────────────────────┤ │
|
||||||
|
│ │ Learn policy: State → Optimal Price │ │
|
||||||
|
│ │ 100k timesteps training │ │
|
||||||
|
│ └────────────────────────────────────────┘ │
|
||||||
|
│ ↓ │
|
||||||
|
│ BENCHMARK vs Baselines: │
|
||||||
|
│ - Fixed pricing: 1.0x always │
|
||||||
|
│ - Simple surge: 1.2x if demand > 10, else 0.9x │
|
||||||
|
│ - Elasticity-based: formula │
|
||||||
|
│ - RL policy: learned │
|
||||||
|
│ - Multi-task + RL: Use MT model predictions as state features │
|
||||||
|
│ │
|
||||||
|
│ VALIDATION: │
|
||||||
|
│ - Calibrate simulator from historical data │
|
||||||
|
│ - Run counterfactuals ("what if agent_frac=0.8?") │
|
||||||
|
│ - A/B test winner on real traffic │
|
||||||
|
│ │
|
||||||
|
└──────────────────────────────────────────────────────────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
## Data Flow (Production)
|
||||||
|
|
||||||
|
```
|
||||||
|
┌─────────────┐
|
||||||
|
│ Browser │
|
||||||
|
│ (User/Agent)│
|
||||||
|
└──────┬──────┘
|
||||||
|
│ POST /api/ingest (events + experimentId)
|
||||||
|
↓
|
||||||
|
┌──────────────┐
|
||||||
|
│ Next.js API │
|
||||||
|
└──────┬───────┘
|
||||||
|
│ Forward events
|
||||||
|
↓
|
||||||
|
┌──────────────┐
|
||||||
|
│ FastAPI │
|
||||||
|
│ /api/kafka │
|
||||||
|
│ /ingest │
|
||||||
|
└──────┬───────┘
|
||||||
|
│ Publish
|
||||||
|
↓
|
||||||
|
┌─────────────────────────┐
|
||||||
|
│ Kafka │
|
||||||
|
│ Topic: user-interactions│
|
||||||
|
└──────┬──────────────────┘
|
||||||
|
│
|
||||||
|
├──────────────────┬──────────────────┐
|
||||||
|
↓ ↓ ↓
|
||||||
|
┌──────────────┐ ┌──────────────┐ ┌──────────────────┐
|
||||||
|
│ Airflow │ │ Real-Time │ │ Kafka Streams │
|
||||||
|
│ (Batch) │ │ Inference │ │ (Feature Cache) │
|
||||||
|
│ │ │ │ │ │
|
||||||
|
│ Daily: │ │ On Price │ │ Rolling window │
|
||||||
|
│ - Retrain │ │ Request: │ │ compute session │
|
||||||
|
│ classifier │ │ - Get session│ │ features, push │
|
||||||
|
│ - Retrain MT │ │ features │ │ to Redis │
|
||||||
|
│ model │ │ - Predict │ │ │
|
||||||
|
│ - Publish to │ │ P(agent) │ │ TTL: 1 hour │
|
||||||
|
│ registry │ │ - Predict │ │ │
|
||||||
|
│ │ │ P(purchase)│ │ │
|
||||||
|
│ │ │ - Compute │ │ │
|
||||||
|
│ │ │ optimal_p │ │ │
|
||||||
|
└──────┬───────┘ └──────┬───────┘ └────────┬─────────┘
|
||||||
|
│ │ │
|
||||||
|
↓ ↓ ↓
|
||||||
|
┌──────────────────────────────────────────────┐
|
||||||
|
│ Redis (Model Registry) │
|
||||||
|
├──────────────────────────────────────────────┤
|
||||||
|
│ Keys: │
|
||||||
|
│ - classifier:agent_detector:latest (pickle) │
|
||||||
|
│ - multitask_model:latest (state_dict) │
|
||||||
|
│ - session_features:{sessionId} (json, TTL) │
|
||||||
|
│ - prices:latest (DataFrame) │
|
||||||
|
│ - elasticity:latest (DataFrame) │
|
||||||
|
└──────────────────┬───────────────────────────┘
|
||||||
|
│
|
||||||
|
↓
|
||||||
|
┌─────────────────────┐
|
||||||
|
│ Pricing Provider │
|
||||||
|
│ /api/{mode}/price/ │
|
||||||
|
│ {productId} │
|
||||||
|
│ │
|
||||||
|
│ GET sessionId │
|
||||||
|
│ → Load features │
|
||||||
|
│ → Load models │
|
||||||
|
│ → Predict │
|
||||||
|
│ → Return price │
|
||||||
|
└─────────┬───────────┘
|
||||||
|
│
|
||||||
|
↓
|
||||||
|
┌─────────────────────┐
|
||||||
|
│ Frontend │
|
||||||
|
│ (Display price) │
|
||||||
|
└─────────────────────┘
|
||||||
|
```
|
||||||
|
|
||||||
|
## Key Metrics
|
||||||
|
|
||||||
|
### Model Performance
|
||||||
|
| Metric | Target | Current | Phase |
|
||||||
|
|--------|--------|---------|-------|
|
||||||
|
| Agent Classifier ROC-AUC | >0.90 | N/A (rule-based) | Phase 2 |
|
||||||
|
| Purchase Predictor ROC-AUC | >0.75 | N/A | Phase 3 |
|
||||||
|
| Pricing Latency (p99) | <100ms | ~50ms | All |
|
||||||
|
| Retraining Frequency | Daily | Every 15min (rules) | Phase 2+ |
|
||||||
|
|
||||||
|
### Business Impact
|
||||||
|
| Metric | Target | Current | Phase |
|
||||||
|
|--------|--------|---------|-------|
|
||||||
|
| Margin Leakage Reduction | -30% | Baseline | Phase 2-4 |
|
||||||
|
| Human Conversion Rate | No change | Baseline | All |
|
||||||
|
| Agent Detection Rate | >85% precision | ~60% (velocity) | Phase 2 |
|
||||||
|
| Revenue Uplift | +10% | Baseline | Phase 3-4 |
|
||||||
|
|
||||||
|
## File Structure (New)
|
||||||
|
|
||||||
|
```
|
||||||
|
experiments/
|
||||||
|
ml/
|
||||||
|
__init__.py
|
||||||
|
|
||||||
|
# Phase 1: Features
|
||||||
|
features/
|
||||||
|
__init__.py
|
||||||
|
temporal.py # TemporalFeatureExtractor
|
||||||
|
behavioral.py # BehavioralFeatureExtractor
|
||||||
|
product.py # ProductFeatureExtractor
|
||||||
|
useragent.py # UserAgentParser
|
||||||
|
aggregator.py # SessionAggregator
|
||||||
|
|
||||||
|
pipeline.py # build_feature_pipeline()
|
||||||
|
datasets.py # load_events_from_kafka(), etc.
|
||||||
|
|
||||||
|
# Phase 2: Classifier
|
||||||
|
train_classifier.py # XGBoost training script
|
||||||
|
|
||||||
|
# Phase 3: Multi-Task
|
||||||
|
models/
|
||||||
|
__init__.py
|
||||||
|
multitask.py # MultiTaskPricingModel (PyTorch)
|
||||||
|
|
||||||
|
train_multitask.py # Multi-task training script
|
||||||
|
distill.py # Knowledge distillation
|
||||||
|
|
||||||
|
# Phase 4: Simulator
|
||||||
|
simulator/
|
||||||
|
__init__.py
|
||||||
|
env.py # DynamicPricingEnv (Gymnasium)
|
||||||
|
agents.py # HumanUser, AgentUser
|
||||||
|
train_rl.py # PPO training
|
||||||
|
|
||||||
|
# Inference
|
||||||
|
inference/
|
||||||
|
__init__.py
|
||||||
|
pricing_service.py # gRPC service (optional)
|
||||||
|
feature_cache.py # Redis feature store client
|
||||||
|
|
||||||
|
# Notebooks
|
||||||
|
notebooks/
|
||||||
|
01_eda.ipynb
|
||||||
|
02_feature_analysis.ipynb
|
||||||
|
03_model_evaluation.ipynb
|
||||||
|
04_simulator_calibration.ipynb
|
||||||
|
```
|
||||||
|
|
||||||
|
## Critical Code Changes
|
||||||
|
|
||||||
|
### 1. Replace Messy SessionState
|
||||||
|
**Before:** `experiments/procesing/steps/session.py` (O(n²) loops)
|
||||||
|
**After:** `experiments/ml/pipeline.py` (vectorized pipeline)
|
||||||
|
|
||||||
|
### 2. Upgrade Pricing Provider
|
||||||
|
**Before:** Simple velocity threshold
|
||||||
|
**After:** ML model inference with agent probability
|
||||||
|
|
||||||
|
### 3. Add Real-Time Feature Store
|
||||||
|
**Before:** No feature caching
|
||||||
|
**After:** Kafka Streams → Redis (session features)
|
||||||
|
|
||||||
|
### 4. Airflow DAG Upgrades
|
||||||
|
**Before:** `surge_pricing_pipeline` (rule-based)
|
||||||
|
**After:** Add `agent_classifier_training_pipeline` (daily retrain)
|
||||||
|
|
||||||
|
## Next Actions (Start Here)
|
||||||
|
|
||||||
|
1. ✅ **Read gameplan**: See `/home/user/PHANTOM/docs/GAMEPLAN_MULTITASK_PRICING.md`
|
||||||
|
|
||||||
|
2. **Create directory structure**:
|
||||||
|
```bash
|
||||||
|
mkdir -p experiments/ml/{features,models,simulator,inference,notebooks}
|
||||||
|
```
|
||||||
|
|
||||||
|
3. **Pull sample data**:
|
||||||
|
```python
|
||||||
|
# experiments/ml/notebooks/01_eda.ipynb
|
||||||
|
from kafka import KafkaConsumer
|
||||||
|
# Pull 1 week of events, join with experiments table
|
||||||
|
# Analyze label distribution, feature correlations
|
||||||
|
```
|
||||||
|
|
||||||
|
4. **Prototype first feature extractor**:
|
||||||
|
```python
|
||||||
|
# experiments/ml/features/temporal.py
|
||||||
|
# Start with TemporalFeatureExtractor
|
||||||
|
# Test on 10k events, validate output schema
|
||||||
|
```
|
||||||
|
|
||||||
|
5. **Review with team**: Discuss tradeoffs, priorities, timeline
|
||||||
|
|
||||||
|
## Questions to Resolve
|
||||||
|
|
||||||
|
1. **Label Quality**: How confident are we in `xp_human_only` labels? Should we add manual verification?
|
||||||
|
|
||||||
|
2. **Compute Budget**: Do we have GPU access for PyTorch training? (Phase 3)
|
||||||
|
|
||||||
|
3. **Latency Requirements**: Is 100ms p99 acceptable for pricing API?
|
||||||
|
|
||||||
|
4. **A/B Testing**: Do we have infrastructure for traffic splitting? (Deployment)
|
||||||
|
|
||||||
|
5. **Monitoring**: Who owns the Grafana dashboards? What alerting thresholds?
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
**For detailed implementation, see:** `/home/user/PHANTOM/docs/GAMEPLAN_MULTITASK_PRICING.md`
|
||||||
1596
docs/GAMEPLAN_MULTITASK_PRICING.md
Normal file
1596
docs/GAMEPLAN_MULTITASK_PRICING.md
Normal file
File diff suppressed because it is too large
Load Diff
@@ -1,115 +0,0 @@
|
|||||||
from airflow import DAG, Dataset
|
|
||||||
from airflow.decorators import task
|
|
||||||
from airflow.utils.dates import days_ago
|
|
||||||
from datetime import timedelta
|
|
||||||
import pandas as pd
|
|
||||||
import logging
|
|
||||||
import sys
|
|
||||||
import pickle
|
|
||||||
|
|
||||||
sys.path.insert(0, '/opt/airflow')
|
|
||||||
|
|
||||||
from procesing.context import PipelineContext
|
|
||||||
from procesing.providers import SupabaseProvider, BackendAPIProvider
|
|
||||||
from procesing.steps import (
|
|
||||||
FetchInteractionsStep,
|
|
||||||
ValidateDataStep,
|
|
||||||
ExtractSessionFeaturesStep,
|
|
||||||
JoinLabelsStep,
|
|
||||||
)
|
|
||||||
|
|
||||||
TRAINING_DATASET = Dataset('phantom://ml/training-data')
|
|
||||||
|
|
||||||
DEFAULT_ARGS = {
|
|
||||||
'owner': 'phantom-research',
|
|
||||||
'depends_on_past': False,
|
|
||||||
'email_on_failure': False,
|
|
||||||
'email_on_retry': False,
|
|
||||||
'retries': 2,
|
|
||||||
'retry_delay': timedelta(minutes=5),
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
class CompositeProvider(SupabaseProvider, BackendAPIProvider):
|
|
||||||
def __init__(self):
|
|
||||||
SupabaseProvider.__init__(self)
|
|
||||||
BackendAPIProvider.__init__(self)
|
|
||||||
|
|
||||||
|
|
||||||
def _get_context(store_mode: str = 'hotel') -> PipelineContext:
|
|
||||||
return PipelineContext(provider=CompositeProvider(), store_mode=store_mode)
|
|
||||||
|
|
||||||
|
|
||||||
with DAG(
|
|
||||||
'ml_training_pipeline',
|
|
||||||
default_args=DEFAULT_ARGS,
|
|
||||||
description='ML training data pipeline: fetch -> validate -> extract features -> label -> publish',
|
|
||||||
schedule=None,
|
|
||||||
start_date=days_ago(1),
|
|
||||||
catchup=False,
|
|
||||||
max_active_runs=1,
|
|
||||||
tags=['ml', 'training', 'features', 'research'],
|
|
||||||
) as dag:
|
|
||||||
|
|
||||||
@task
|
|
||||||
def fetch_interactions(**kwargs) -> bytes:
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
|
|
||||||
df = FetchInteractionsStep(ctx).transform(None)
|
|
||||||
logging.info(f"Fetched {len(df)} interactions, {df['sessionId'].nunique()} sessions")
|
|
||||||
return pickle.dumps(df)
|
|
||||||
|
|
||||||
@task
|
|
||||||
def validate_data(raw_data: bytes, **kwargs) -> bytes:
|
|
||||||
df = pickle.loads(raw_data)
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
|
|
||||||
validated = ValidateDataStep(ctx).transform(df)
|
|
||||||
report = ctx.get_cached('validation_report') or {}
|
|
||||||
logging.info(f"Validation: {report.get('status')}, {report.get('sessions', 0)} sessions")
|
|
||||||
return pickle.dumps(validated)
|
|
||||||
|
|
||||||
@task
|
|
||||||
def extract_session_features(validated_data: bytes, **kwargs) -> bytes:
|
|
||||||
df = pickle.loads(validated_data)
|
|
||||||
if df.empty:
|
|
||||||
logging.warning("Empty input, skipping feature extraction")
|
|
||||||
return pickle.dumps(pd.DataFrame())
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
|
|
||||||
features = ExtractSessionFeaturesStep(ctx).transform(df)
|
|
||||||
logging.info(f"Extracted {len(features.columns)} features for {len(features)} sessions")
|
|
||||||
return pickle.dumps(features)
|
|
||||||
|
|
||||||
@task
|
|
||||||
def join_labels(features_data: bytes, **kwargs) -> bytes:
|
|
||||||
features_df = pickle.loads(features_data)
|
|
||||||
if features_df.empty:
|
|
||||||
logging.warning("Empty features, skipping label join")
|
|
||||||
return pickle.dumps(pd.DataFrame())
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
ctx = _get_context(dag_conf.get('store_mode', 'hotel'))
|
|
||||||
labeled = JoinLabelsStep(ctx).transform(features_df)
|
|
||||||
n_agents = labeled['is_agent'].sum() if 'is_agent' in labeled.columns else 0
|
|
||||||
logging.info(f"Labeled {len(labeled)} sessions: {n_agents} agents")
|
|
||||||
return pickle.dumps(labeled)
|
|
||||||
|
|
||||||
@task(outlets=[TRAINING_DATASET])
|
|
||||||
def publish_training_data(labeled_data: bytes, **kwargs) -> dict:
|
|
||||||
labeled_df = pickle.loads(labeled_data)
|
|
||||||
if labeled_df.empty:
|
|
||||||
return {'status': 'skipped', 'reason': 'empty_data'}
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
return {
|
|
||||||
'status': 'success',
|
|
||||||
'n_sessions': len(labeled_df),
|
|
||||||
'n_features': len([c for c in labeled_df.columns if c not in ['sessionId', 'experimentId', 'is_agent']]),
|
|
||||||
'store_mode': dag_conf.get('store_mode', 'hotel'),
|
|
||||||
'timestamp': pd.Timestamp.now().isoformat(),
|
|
||||||
}
|
|
||||||
|
|
||||||
raw = fetch_interactions()
|
|
||||||
validated = validate_data(raw)
|
|
||||||
features = extract_session_features(validated)
|
|
||||||
labeled = join_labels(features)
|
|
||||||
publish_training_data(labeled)
|
|
||||||
@@ -1,210 +0,0 @@
|
|||||||
from airflow import DAG
|
|
||||||
from airflow.operators.python import PythonOperator
|
|
||||||
from airflow.utils.dates import days_ago
|
|
||||||
from datetime import timedelta
|
|
||||||
import pandas as pd
|
|
||||||
import logging
|
|
||||||
import sys
|
|
||||||
import pickle
|
|
||||||
|
|
||||||
sys.path.insert(0, '/opt/airflow')
|
|
||||||
|
|
||||||
from procesing.context import PipelineContext
|
|
||||||
from procesing.providers import SupabaseProvider, BackendAPIProvider
|
|
||||||
from procesing.steps import (
|
|
||||||
FetchInteractionsStep,
|
|
||||||
FetchPriceLogsStep,
|
|
||||||
ComputeDemandStep,
|
|
||||||
AggregatePriceLogsStep,
|
|
||||||
JoinProductFeaturesStep,
|
|
||||||
)
|
|
||||||
from procesing.pricers.simple import SimpleSurgePricer
|
|
||||||
|
|
||||||
DEFAULT_ARGS = {
|
|
||||||
'owner': 'phantom-research',
|
|
||||||
'depends_on_past': False,
|
|
||||||
'email_on_failure': False,
|
|
||||||
'email_on_retry': False,
|
|
||||||
'retries': 2,
|
|
||||||
'retry_delay': timedelta(minutes=5),
|
|
||||||
}
|
|
||||||
|
|
||||||
class CompositeProvider(SupabaseProvider, BackendAPIProvider):
|
|
||||||
def __init__(self):
|
|
||||||
SupabaseProvider.__init__(self)
|
|
||||||
BackendAPIProvider.__init__(self)
|
|
||||||
|
|
||||||
def _get_provider():
|
|
||||||
return CompositeProvider()
|
|
||||||
|
|
||||||
def _make_task_callables(store_mode: str):
|
|
||||||
"""Generate task callables bound to a specific store_mode."""
|
|
||||||
|
|
||||||
def get_context(**kwargs):
|
|
||||||
return PipelineContext(provider=_get_provider(), store_mode=store_mode)
|
|
||||||
|
|
||||||
def fetch_interactions(**kwargs):
|
|
||||||
ctx = get_context(**kwargs)
|
|
||||||
df = FetchInteractionsStep(ctx).transform(None)
|
|
||||||
kwargs['ti'].xcom_push(key='interactions_raw', value=pickle.dumps(df))
|
|
||||||
logging.info(f"[{store_mode}] Fetched {len(df)} interaction records")
|
|
||||||
return len(df)
|
|
||||||
|
|
||||||
def fetch_price_logs(**kwargs):
|
|
||||||
ctx = get_context(**kwargs)
|
|
||||||
df = FetchPriceLogsStep(ctx).transform(None)
|
|
||||||
kwargs['ti'].xcom_push(key='price_logs_raw', value=pickle.dumps(df))
|
|
||||||
logging.info(f"[{store_mode}] Fetched {len(df)} price records")
|
|
||||||
return len(df)
|
|
||||||
|
|
||||||
def compute_demand(**kwargs):
|
|
||||||
ti = kwargs['ti']
|
|
||||||
df = pickle.loads(ti.xcom_pull(key='interactions_raw'))
|
|
||||||
ctx = get_context(**kwargs)
|
|
||||||
demand_df = ComputeDemandStep(ctx).transform(df)
|
|
||||||
ti.xcom_push(key='demand_data', value=pickle.dumps(demand_df))
|
|
||||||
logging.info(f"[{store_mode}] Computed demand for {len(demand_df)} products")
|
|
||||||
return len(demand_df)
|
|
||||||
|
|
||||||
def aggregate_price_logs(**kwargs):
|
|
||||||
ti = kwargs['ti']
|
|
||||||
df = pickle.loads(ti.xcom_pull(key='price_logs_raw'))
|
|
||||||
ctx = get_context(**kwargs)
|
|
||||||
price_df = AggregatePriceLogsStep(ctx).transform(df)
|
|
||||||
ti.xcom_push(key='price_data', value=pickle.dumps(price_df))
|
|
||||||
logging.info(f"[{store_mode}] Aggregated price logs for {len(price_df)} products")
|
|
||||||
return len(price_df)
|
|
||||||
|
|
||||||
def join_product_features(**kwargs):
|
|
||||||
ti = kwargs['ti']
|
|
||||||
demand_df = pickle.loads(ti.xcom_pull(key='demand_data'))
|
|
||||||
price_df = pickle.loads(ti.xcom_pull(key='price_data'))
|
|
||||||
ctx = get_context(**kwargs)
|
|
||||||
joined_df = JoinProductFeaturesStep(ctx).transform((demand_df, price_df))
|
|
||||||
ti.xcom_push(key='product_features', value=pickle.dumps(joined_df))
|
|
||||||
logging.info(f"[{store_mode}] Joined features for {len(joined_df)} products")
|
|
||||||
return len(joined_df)
|
|
||||||
|
|
||||||
def apply_surge_pricing(**kwargs):
|
|
||||||
ti = kwargs['ti']
|
|
||||||
product_features = pickle.loads(ti.xcom_pull(key='product_features'))
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
|
|
||||||
data = product_features.rename(columns={'demand_score': 'demand'})
|
|
||||||
surge_pricer = SimpleSurgePricer(
|
|
||||||
high_threshold=dag_conf.get('high_threshold', 10),
|
|
||||||
low_threshold=dag_conf.get('low_threshold', 2),
|
|
||||||
surge_multiplier=dag_conf.get('surge_multiplier', 1.2),
|
|
||||||
discount_multiplier=dag_conf.get('discount_multiplier', 0.9)
|
|
||||||
)
|
|
||||||
surge_pricer.fit(data)
|
|
||||||
data['optimal_price'] = surge_pricer.predict()
|
|
||||||
|
|
||||||
prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
|
|
||||||
'price': 'current_price', 'demand': 'demand_score'
|
|
||||||
})
|
|
||||||
ti.xcom_push(key='predicted_prices', value=pickle.dumps(prices_df))
|
|
||||||
logging.info(f"[{store_mode}] Applied surge pricing for {len(prices_df)} products")
|
|
||||||
return len(prices_df)
|
|
||||||
|
|
||||||
def publish_results(**kwargs):
|
|
||||||
ti = kwargs['ti']
|
|
||||||
prices_df = pickle.loads(ti.xcom_pull(key='predicted_prices'))
|
|
||||||
from lib.model_registry import ModelRegistry
|
|
||||||
|
|
||||||
registry = ModelRegistry()
|
|
||||||
dag_conf = kwargs.get('dag_run').conf if kwargs.get('dag_run') else {}
|
|
||||||
|
|
||||||
metadata = {
|
|
||||||
'timestamp': pd.Timestamp.now().isoformat(),
|
|
||||||
'store_mode': store_mode,
|
|
||||||
'dag_run_id': kwargs['dag_run'].run_id if kwargs.get('dag_run') else 'manual',
|
|
||||||
'pricing_method': 'surge',
|
|
||||||
'high_threshold': dag_conf.get('high_threshold', 10),
|
|
||||||
'low_threshold': dag_conf.get('low_threshold', 2),
|
|
||||||
'surge_multiplier': dag_conf.get('surge_multiplier', 1.2),
|
|
||||||
'discount_multiplier': dag_conf.get('discount_multiplier', 0.9)
|
|
||||||
}
|
|
||||||
registry.publish_prices(prices_df, model_name=f'{store_mode}_latest', metadata=metadata)
|
|
||||||
logging.info(f"[{store_mode}] Published surge pricing for {len(prices_df)} products")
|
|
||||||
|
|
||||||
return {
|
|
||||||
'n_products': len(prices_df),
|
|
||||||
'registry_status': 'success',
|
|
||||||
'store_mode': store_mode,
|
|
||||||
'mean_demand': float(prices_df['demand_score'].mean()) if 'demand_score' in prices_df.columns else None
|
|
||||||
}
|
|
||||||
|
|
||||||
return {
|
|
||||||
'fetch_interactions': fetch_interactions,
|
|
||||||
'fetch_price_logs': fetch_price_logs,
|
|
||||||
'compute_demand': compute_demand,
|
|
||||||
'aggregate_price_logs': aggregate_price_logs,
|
|
||||||
'join_product_features': join_product_features,
|
|
||||||
'apply_surge_pricing': apply_surge_pricing,
|
|
||||||
'publish_results': publish_results,
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
def create_surge_pricing_dag(store_mode: str) -> DAG:
|
|
||||||
"""Factory: generates a surge pricing DAG for a given store_mode."""
|
|
||||||
callables = _make_task_callables(store_mode)
|
|
||||||
|
|
||||||
dag = DAG(
|
|
||||||
f'surge_pricing_{store_mode}',
|
|
||||||
default_args=DEFAULT_ARGS,
|
|
||||||
description=f'Surge pricing pipeline for {store_mode} store mode',
|
|
||||||
schedule_interval='*/15 * * * *',
|
|
||||||
start_date=days_ago(1),
|
|
||||||
catchup=False,
|
|
||||||
max_active_runs=1,
|
|
||||||
tags=['pricing', 'surge', 'research', store_mode],
|
|
||||||
)
|
|
||||||
|
|
||||||
with dag:
|
|
||||||
t_fetch_interactions = PythonOperator(
|
|
||||||
task_id='fetch_interactions',
|
|
||||||
python_callable=callables['fetch_interactions'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_fetch_price_logs = PythonOperator(
|
|
||||||
task_id='fetch_price_logs',
|
|
||||||
python_callable=callables['fetch_price_logs'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_compute_demand = PythonOperator(
|
|
||||||
task_id='compute_demand',
|
|
||||||
python_callable=callables['compute_demand'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_aggregate_prices = PythonOperator(
|
|
||||||
task_id='aggregate_price_logs',
|
|
||||||
python_callable=callables['aggregate_price_logs'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_join_features = PythonOperator(
|
|
||||||
task_id='join_product_features',
|
|
||||||
python_callable=callables['join_product_features'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_surge_pricing = PythonOperator(
|
|
||||||
task_id='apply_surge_pricing',
|
|
||||||
python_callable=callables['apply_surge_pricing'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
t_publish = PythonOperator(
|
|
||||||
task_id='publish_results',
|
|
||||||
python_callable=callables['publish_results'],
|
|
||||||
provide_context=True,
|
|
||||||
)
|
|
||||||
|
|
||||||
t_fetch_interactions >> t_compute_demand
|
|
||||||
t_fetch_price_logs >> t_aggregate_prices
|
|
||||||
[t_compute_demand, t_aggregate_prices] >> t_join_features >> t_surge_pricing >> t_publish
|
|
||||||
|
|
||||||
return dag
|
|
||||||
|
|
||||||
|
|
||||||
# instantiate DAGs for Airflow to discover
|
|
||||||
dag_airline = create_surge_pricing_dag('airline')
|
|
||||||
dag_hotel = create_surge_pricing_dag('hotel')
|
|
||||||
@@ -1,11 +0,0 @@
|
|||||||
from .evals import evaluate
|
|
||||||
from .arch import (
|
|
||||||
XGBoostAgentClassifier,
|
|
||||||
LightGBMAgentClassifier
|
|
||||||
)
|
|
||||||
|
|
||||||
__all__ =[
|
|
||||||
'evaluate',
|
|
||||||
'XGBoostAgentClassifier',
|
|
||||||
'LightGBMAgentClassifier'
|
|
||||||
]
|
|
||||||
@@ -1,122 +0,0 @@
|
|||||||
# sklearn compatible models for agent detection
|
|
||||||
from sklearn.base import BaseEstimator, ClassifierMixin
|
|
||||||
from procesing.context import PipelineContext
|
|
||||||
from typing import Any, Optional, Tuple
|
|
||||||
from abc import ABC, abstractmethod
|
|
||||||
import xgboost as xgb
|
|
||||||
import lightgbm as lgb
|
|
||||||
import numpy as np
|
|
||||||
import pandas as pd
|
|
||||||
|
|
||||||
TASK = 'classification'
|
|
||||||
LABELS = ['human', 'agent']
|
|
||||||
|
|
||||||
|
|
||||||
class BaseAgentClassifier(BaseEstimator, ClassifierMixin, ABC):
|
|
||||||
"""Base class for tree-based agent detection classifiers with common logic"""
|
|
||||||
|
|
||||||
def __init__(self, context: Optional[PipelineContext] = None, n_estimators: int = 200,
|
|
||||||
max_depth: int = 6, learning_rate: float = 0.05,
|
|
||||||
early_stopping_rounds: int = 20):
|
|
||||||
self.context = context
|
|
||||||
self.n_estimators = n_estimators
|
|
||||||
self.max_depth = max_depth
|
|
||||||
self.learning_rate = learning_rate
|
|
||||||
self.early_stopping_rounds = early_stopping_rounds
|
|
||||||
self.model_ = None
|
|
||||||
self.feature_names_ = None
|
|
||||||
|
|
||||||
def _to_array(self, X):
|
|
||||||
"""Convert pandas structures to numpy arrays"""
|
|
||||||
return X.values if isinstance(X, (pd.DataFrame, pd.Series)) else X
|
|
||||||
|
|
||||||
def _compute_pos_weight(self, y_arr):
|
|
||||||
"""Calculate scale_pos_weight for class imbalance handling"""
|
|
||||||
n_neg, n_pos = (y_arr == 0).sum(), (y_arr == 1).sum()
|
|
||||||
return n_neg / n_pos if n_pos > 0 else 1.0
|
|
||||||
|
|
||||||
def _prepare_eval_set(self, eval_set):
|
|
||||||
"""Convert eval_set to numpy arrays if needed"""
|
|
||||||
if not eval_set:
|
|
||||||
return None
|
|
||||||
X_val, y_val = eval_set[0]
|
|
||||||
return [(self._to_array(X_val), self._to_array(y_val))]
|
|
||||||
|
|
||||||
@abstractmethod
|
|
||||||
def _build_model(self, scale_pos: float):
|
|
||||||
"""Build the underlying model instance (must be implemented by subclasses)"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
@abstractmethod
|
|
||||||
def _fit_with_eval(self, X_arr, y_arr, eval_arr):
|
|
||||||
"""Fit model with evaluation set (must be implemented by subclasses)"""
|
|
||||||
pass
|
|
||||||
|
|
||||||
def fit(self, X, y, eval_set=None):
|
|
||||||
X_arr, y_arr = self._to_array(X), self._to_array(y)
|
|
||||||
|
|
||||||
if isinstance(X, pd.DataFrame):
|
|
||||||
self.feature_names_ = X.columns.tolist()
|
|
||||||
|
|
||||||
scale_pos = self._compute_pos_weight(y_arr)
|
|
||||||
self.model_ = self._build_model(scale_pos)
|
|
||||||
|
|
||||||
eval_arr = self._prepare_eval_set(eval_set)
|
|
||||||
if eval_arr:
|
|
||||||
self._fit_with_eval(X_arr, y_arr, eval_arr)
|
|
||||||
else:
|
|
||||||
self.model_.fit(X_arr, y_arr)
|
|
||||||
|
|
||||||
return self
|
|
||||||
|
|
||||||
def predict(self, X):
|
|
||||||
return self.model_.predict(self._to_array(X))
|
|
||||||
|
|
||||||
def predict_proba(self, X):
|
|
||||||
return self.model_.predict_proba(self._to_array(X))
|
|
||||||
|
|
||||||
@property
|
|
||||||
def feature_importances_(self):
|
|
||||||
return self.model_.feature_importances_ if self.model_ else None
|
|
||||||
|
|
||||||
|
|
||||||
class XGBoostAgentClassifier(BaseAgentClassifier):
|
|
||||||
"""XGBoost binary classifier for agent detection with class imbalance handling"""
|
|
||||||
|
|
||||||
def _build_model(self, scale_pos: float):
|
|
||||||
return xgb.XGBClassifier(
|
|
||||||
n_estimators=self.n_estimators,
|
|
||||||
max_depth=self.max_depth,
|
|
||||||
learning_rate=self.learning_rate,
|
|
||||||
scale_pos_weight=scale_pos,
|
|
||||||
eval_metric='auc',
|
|
||||||
early_stopping_rounds=self.early_stopping_rounds,
|
|
||||||
random_state=42,
|
|
||||||
tree_method='hist',
|
|
||||||
enable_categorical=False
|
|
||||||
)
|
|
||||||
|
|
||||||
def _fit_with_eval(self, X_arr, y_arr, eval_arr):
|
|
||||||
self.model_.fit(X_arr, y_arr, eval_set=eval_arr, verbose=False)
|
|
||||||
|
|
||||||
|
|
||||||
class LightGBMAgentClassifier(BaseAgentClassifier):
|
|
||||||
"""LightGBM binary classifier for agent detection with class imbalance handling"""
|
|
||||||
|
|
||||||
def _build_model(self, scale_pos: float):
|
|
||||||
return lgb.LGBMClassifier(
|
|
||||||
n_estimators=self.n_estimators,
|
|
||||||
max_depth=self.max_depth,
|
|
||||||
learning_rate=self.learning_rate,
|
|
||||||
scale_pos_weight=scale_pos,
|
|
||||||
metric='auc',
|
|
||||||
random_state=42,
|
|
||||||
verbosity=-1
|
|
||||||
)
|
|
||||||
|
|
||||||
def _fit_with_eval(self, X_arr, y_arr, eval_arr):
|
|
||||||
self.model_.fit(
|
|
||||||
X_arr, y_arr,
|
|
||||||
eval_set=eval_arr,
|
|
||||||
callbacks=[lgb.early_stopping(self.early_stopping_rounds, verbose=False)]
|
|
||||||
)
|
|
||||||
@@ -1,103 +0,0 @@
|
|||||||
from sklearn.metrics import (accuracy_score, precision_score, recall_score,
|
|
||||||
f1_score, roc_auc_score, confusion_matrix, roc_curve)
|
|
||||||
from torch.utils.tensorboard import SummaryWriter
|
|
||||||
from logging import getLogger
|
|
||||||
import numpy as np
|
|
||||||
import matplotlib.pyplot as plt
|
|
||||||
import io
|
|
||||||
from PIL import Image
|
|
||||||
|
|
||||||
logger = getLogger(__name__)
|
|
||||||
|
|
||||||
|
|
||||||
def log_feature_importance(writer, model, feature_names, epoch):
|
|
||||||
"""Visualize and log feature importance to TensorBoard"""
|
|
||||||
if not hasattr(model, 'feature_importances_') or model.feature_importances_ is None:
|
|
||||||
return
|
|
||||||
|
|
||||||
importance = model.feature_importances_
|
|
||||||
indices = np.argsort(importance)[::-1][:20] # top 20
|
|
||||||
top_features = [feature_names[i] for i in indices]
|
|
||||||
top_importance = importance[indices]
|
|
||||||
|
|
||||||
for i, (feat, imp) in enumerate(zip(top_features, top_importance)):
|
|
||||||
writer.add_scalar(f'FeatureImportance/{feat}', imp, epoch)
|
|
||||||
|
|
||||||
fig, ax = plt.subplots(figsize=(10, 8))
|
|
||||||
ax.barh(range(len(top_features)), top_importance, align='center')
|
|
||||||
ax.set_yticks(range(len(top_features)))
|
|
||||||
ax.set_yticklabels(top_features)
|
|
||||||
ax.invert_yaxis()
|
|
||||||
ax.set_xlabel('Importance')
|
|
||||||
ax.set_title(f'Top 20 Feature Importance (Epoch {epoch})')
|
|
||||||
ax.grid(axis='x', alpha=0.3)
|
|
||||||
|
|
||||||
buf = io.BytesIO()
|
|
||||||
plt.tight_layout()
|
|
||||||
plt.savefig(buf, format='png', dpi=100)
|
|
||||||
buf.seek(0)
|
|
||||||
img = Image.open(buf)
|
|
||||||
img_arr = np.array(img)
|
|
||||||
writer.add_image('FeatureImportance/Chart', img_arr, epoch, dataformats='HWC')
|
|
||||||
plt.close()
|
|
||||||
|
|
||||||
def evaluate(perdicted_class, predicted_proba, true_class, writer: SummaryWriter, epoch: int):
|
|
||||||
accuracy = accuracy_score(true_class, perdicted_class)
|
|
||||||
precision = precision_score(true_class, perdicted_class, zero_division=0)
|
|
||||||
recall = recall_score(true_class, perdicted_class, zero_division=0)
|
|
||||||
f1 = f1_score(true_class, perdicted_class, zero_division=0)
|
|
||||||
roc_auc = roc_auc_score(true_class, predicted_proba)
|
|
||||||
|
|
||||||
writer.add_scalar('Eval/Accuracy', accuracy, epoch)
|
|
||||||
writer.add_scalar('Eval/Precision', precision, epoch)
|
|
||||||
writer.add_scalar('Eval/Recall', recall, epoch)
|
|
||||||
writer.add_scalar('Eval/F1_Score', f1, epoch)
|
|
||||||
writer.add_scalar('Eval/ROC_AUC', roc_auc, epoch)
|
|
||||||
|
|
||||||
# confusion matrix
|
|
||||||
cm = confusion_matrix(true_class, perdicted_class)
|
|
||||||
tn, fp, fn, tp = cm.ravel()
|
|
||||||
writer.add_scalar('Eval/TrueNeg', tn, epoch)
|
|
||||||
writer.add_scalar('Eval/FalsePos', fp, epoch)
|
|
||||||
writer.add_scalar('Eval/FalseNeg', fn, epoch)
|
|
||||||
writer.add_scalar('Eval/TruePos', tp, epoch)
|
|
||||||
|
|
||||||
# specificity and sensitivity
|
|
||||||
specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
|
|
||||||
sensitivity = recall # same as recall/TPR
|
|
||||||
writer.add_scalar('Eval/Specificity', specificity, epoch)
|
|
||||||
writer.add_scalar('Eval/Sensitivity', sensitivity, epoch)
|
|
||||||
|
|
||||||
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(12, 4))
|
|
||||||
ax1.matshow(cm, cmap='Blues', alpha=0.7)
|
|
||||||
for i in range(2):
|
|
||||||
for j in range(2):
|
|
||||||
ax1.text(j, i, str(cm[i, j]), ha='center', va='center', fontsize=14)
|
|
||||||
ax1.set_xlabel('Predicted')
|
|
||||||
ax1.set_ylabel('True')
|
|
||||||
ax1.set_title(f'Confusion Matrix (Epoch {epoch})')
|
|
||||||
ax1.set_xticks([0, 1])
|
|
||||||
ax1.set_yticks([0, 1])
|
|
||||||
ax1.set_xticklabels(['Human', 'Agent'])
|
|
||||||
ax1.set_yticklabels(['Human', 'Agent'])
|
|
||||||
|
|
||||||
# ROC curve
|
|
||||||
fpr, tpr, _ = roc_curve(true_class, predicted_proba)
|
|
||||||
ax2.plot(fpr, tpr, label=f'AUC={roc_auc:.3f}', linewidth=2)
|
|
||||||
ax2.plot([0, 1], [0, 1], 'k--', label='Random')
|
|
||||||
ax2.set_xlabel('False Positive Rate')
|
|
||||||
ax2.set_ylabel('True Positive Rate')
|
|
||||||
ax2.set_title('ROC Curve')
|
|
||||||
ax2.legend()
|
|
||||||
ax2.grid(alpha=0.3)
|
|
||||||
|
|
||||||
buf = io.BytesIO()
|
|
||||||
plt.tight_layout()
|
|
||||||
plt.savefig(buf, format='png', dpi=100)
|
|
||||||
buf.seek(0)
|
|
||||||
img = Image.open(buf)
|
|
||||||
img_arr = np.array(img)
|
|
||||||
writer.add_image('Eval/Metrics', img_arr, epoch, dataformats='HWC')
|
|
||||||
plt.close()
|
|
||||||
|
|
||||||
logger.info(f"Eval {epoch}: Acc={accuracy:.4f} Prec={precision:.4f} Rec={recall:.4f} F1={f1:.4f} AUC={roc_auc:.4f}")
|
|
||||||
@@ -1,6 +0,0 @@
|
|||||||
torch
|
|
||||||
tensorboard
|
|
||||||
fastparquet
|
|
||||||
pyarrow
|
|
||||||
xgboost
|
|
||||||
lightgbm
|
|
||||||
@@ -1,137 +0,0 @@
|
|||||||
from torch.utils.tensorboard import SummaryWriter
|
|
||||||
from sklearn.model_selection import train_test_split
|
|
||||||
from logging import getLogger
|
|
||||||
from pathlib import Path
|
|
||||||
import pandas as pd
|
|
||||||
import numpy as np
|
|
||||||
import joblib
|
|
||||||
from datetime import datetime
|
|
||||||
from ml.evals import evaluate, log_feature_importance
|
|
||||||
from ml.arch import XGBoostAgentClassifier, LightGBMAgentClassifier, LABELS
|
|
||||||
|
|
||||||
logger = getLogger(__name__)
|
|
||||||
|
|
||||||
FEATURE_COLS_EXCLUDE = ['sessionId', 'experimentId', 'is_agent', 'xp_human_only', 'xp_market_mode', 'browser_family']
|
|
||||||
RUNS_DIR = Path('ml/runs')
|
|
||||||
CHECKPOINTS_DIR = Path('ml/checkpoints')
|
|
||||||
|
|
||||||
|
|
||||||
def prepare_data(df):
|
|
||||||
"""
|
|
||||||
Prepare feature matrix and labels from raw dataframe
|
|
||||||
Handles missing labels, feature selection, and categorical encoding
|
|
||||||
Returns: (X, y, feature_cols)
|
|
||||||
"""
|
|
||||||
# drop rows with missing labels
|
|
||||||
n_before = len(df)
|
|
||||||
df = df[df['is_agent'].notna()].copy()
|
|
||||||
n_dropped = n_before - len(df)
|
|
||||||
if n_dropped > 0:
|
|
||||||
logger.warning(f"Dropped {n_dropped} sessions with missing labels")
|
|
||||||
|
|
||||||
if len(df) == 0:
|
|
||||||
logger.error("No labeled data available")
|
|
||||||
return None, None, None
|
|
||||||
|
|
||||||
feature_cols = [c for c in df.columns if c not in FEATURE_COLS_EXCLUDE]
|
|
||||||
|
|
||||||
# handle categorical browser_family via one-hot encoding
|
|
||||||
if 'browser_family' in df.columns:
|
|
||||||
browser_dummies = pd.get_dummies(df['browser_family'], prefix='browser', drop_first=True)
|
|
||||||
df = pd.concat([df, browser_dummies], axis=1)
|
|
||||||
feature_cols.extend(browser_dummies.columns.tolist())
|
|
||||||
|
|
||||||
X = df[feature_cols].fillna(0)
|
|
||||||
y = df['is_agent'].astype(int)
|
|
||||||
|
|
||||||
return X, y, feature_cols
|
|
||||||
|
|
||||||
|
|
||||||
def train(data_path=None, model_type='xgboost', test_size=0.2, random_state=42,
|
|
||||||
n_estimators=200, max_depth=6, learning_rate=0.05):
|
|
||||||
"""
|
|
||||||
Train agent detection classifier
|
|
||||||
Args:
|
|
||||||
data_path: path to labeled feature matrix CSV or parquet
|
|
||||||
model_type: 'xgboost' or 'lightgbm'
|
|
||||||
test_size: fraction for test split
|
|
||||||
random_state: seed for reproducibility
|
|
||||||
"""
|
|
||||||
RUNS_DIR.mkdir(exist_ok=True)
|
|
||||||
CHECKPOINTS_DIR.mkdir(exist_ok=True)
|
|
||||||
|
|
||||||
run_name = f"{model_type}_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
|
||||||
writer = SummaryWriter(log_dir=RUNS_DIR / run_name)
|
|
||||||
logger.info(f"Starting training run: {run_name}")
|
|
||||||
|
|
||||||
# load data
|
|
||||||
if data_path is None:
|
|
||||||
logger.error("data_path required")
|
|
||||||
return
|
|
||||||
df = pd.read_parquet(data_path)
|
|
||||||
logger.info(f"Loaded {len(df)} sessions from {data_path}")
|
|
||||||
|
|
||||||
# prepare features and labels
|
|
||||||
if 'is_agent' not in df.columns:
|
|
||||||
logger.error("Missing is_agent column")
|
|
||||||
return
|
|
||||||
|
|
||||||
X, y, feature_cols = prepare_data(df)
|
|
||||||
if X is None:
|
|
||||||
return
|
|
||||||
|
|
||||||
# class distribution
|
|
||||||
n_agents = y.sum()
|
|
||||||
n_humans = (y == 0).sum()
|
|
||||||
logger.info(f"Class distribution: {n_humans} humans, {n_agents} agents" + (f" (ratio {n_humans / n_agents:.2f})" if n_agents > 0 else ""))
|
|
||||||
|
|
||||||
# train/test split with stratification
|
|
||||||
X_train, X_test, y_train, y_test = train_test_split(
|
|
||||||
X, y, test_size=test_size, random_state=random_state, stratify=y
|
|
||||||
)
|
|
||||||
logger.info(f"Train: {len(X_train)}, Test: {len(X_test)}")
|
|
||||||
|
|
||||||
# init model
|
|
||||||
if model_type == 'xgboost':
|
|
||||||
model = XGBoostAgentClassifier(
|
|
||||||
n_estimators=n_estimators,
|
|
||||||
max_depth=max_depth,
|
|
||||||
learning_rate=learning_rate
|
|
||||||
)
|
|
||||||
elif model_type == 'lightgbm':
|
|
||||||
model = LightGBMAgentClassifier(
|
|
||||||
n_estimators=n_estimators,
|
|
||||||
max_depth=max_depth,
|
|
||||||
learning_rate=learning_rate
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
logger.error(f"Unknown model type: {model_type}")
|
|
||||||
return
|
|
||||||
|
|
||||||
# train with eval set for early stopping
|
|
||||||
model.fit(X_train, y_train, eval_set=[(X_test, y_test)])
|
|
||||||
logger.info("Training complete")
|
|
||||||
|
|
||||||
# evaluate on test set
|
|
||||||
y_pred = model.predict(X_test)
|
|
||||||
y_prob = model.predict_proba(X_test)[:, 1]
|
|
||||||
|
|
||||||
evaluate(y_pred, y_prob, y_test, writer, epoch=0)
|
|
||||||
|
|
||||||
# log feature importance
|
|
||||||
log_feature_importance(writer, model, X.columns.tolist(), epoch=0)
|
|
||||||
|
|
||||||
# save model
|
|
||||||
model_path = CHECKPOINTS_DIR / f"{run_name}.pkl"
|
|
||||||
joblib.dump({'model': model, 'feature_cols': X.columns.tolist(), 'run_name': run_name}, model_path)
|
|
||||||
logger.info(f"Model saved to {model_path}")
|
|
||||||
|
|
||||||
writer.close()
|
|
||||||
return model, X.columns.tolist()
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
import sys
|
|
||||||
data_path = sys.argv[1]
|
|
||||||
model_type = sys.argv[2] if len(sys.argv) > 2 else 'xgboost'
|
|
||||||
train(data_path, model_type=model_type)
|
|
||||||
@@ -2,7 +2,6 @@ from sklearn.pipeline import Pipeline
|
|||||||
import pandas as pd
|
import pandas as pd
|
||||||
from procesing.context import PipelineContext
|
from procesing.context import PipelineContext
|
||||||
from procesing.providers import SupabaseProvider, BackendAPIProvider
|
from procesing.providers import SupabaseProvider, BackendAPIProvider
|
||||||
import os
|
|
||||||
from procesing.steps import (
|
from procesing.steps import (
|
||||||
FetchInteractionsStep,
|
FetchInteractionsStep,
|
||||||
FetchPriceLogsStep,
|
FetchPriceLogsStep,
|
||||||
@@ -13,13 +12,11 @@ from procesing.steps import (
|
|||||||
ChunkByTimeWindowStep,
|
ChunkByTimeWindowStep,
|
||||||
ComputeDemandForChunksStep,
|
ComputeDemandForChunksStep,
|
||||||
AggregatePriceLogsStep,
|
AggregatePriceLogsStep,
|
||||||
|
# BuildStateSpaceStep,
|
||||||
FitPricingFunctionStep,
|
FitPricingFunctionStep,
|
||||||
PredictPricesStep,
|
PredictPricesStep,
|
||||||
ComputeDemandStep,
|
ComputeDemandStep,
|
||||||
JoinProductFeaturesStep,
|
JoinProductFeaturesStep
|
||||||
ExtractSessionFeaturesStep,
|
|
||||||
JoinLabelsStep,
|
|
||||||
ValidateDataStep,
|
|
||||||
)
|
)
|
||||||
from procesing.pricers import SimpleSurgePricer
|
from procesing.pricers import SimpleSurgePricer
|
||||||
|
|
||||||
@@ -109,66 +106,33 @@ def full_pipeline(context: PipelineContext,
|
|||||||
return product_features_df, optimal_prices_df
|
return product_features_df, optimal_prices_df
|
||||||
|
|
||||||
|
|
||||||
def ml_training_pipeline(context: PipelineContext) -> pd.DataFrame:
|
|
||||||
"""
|
|
||||||
Build labeled session-level feature matrix for ML model training.
|
|
||||||
Pipeline: fetch -> validate -> extract features -> join labels
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
DataFrame with ~25 features per session + is_agent label
|
|
||||||
Columns: sessionId, experimentId, temporal/behavioral/product/ua features, is_agent
|
|
||||||
"""
|
|
||||||
# fetch raw interactions
|
|
||||||
interactions_df = FetchInteractionsStep(context).transform(None)
|
|
||||||
|
|
||||||
# validate data quality (report cached in context)
|
|
||||||
interactions_df = ValidateDataStep(context).transform(interactions_df)
|
|
||||||
if interactions_df.empty:
|
|
||||||
return pd.DataFrame()
|
|
||||||
|
|
||||||
# extract vectorized session features
|
|
||||||
features_df = ExtractSessionFeaturesStep(context).transform(interactions_df)
|
|
||||||
if features_df.empty:
|
|
||||||
return pd.DataFrame()
|
|
||||||
|
|
||||||
# join experiment labels (is_agent = ~xp_human_only)
|
|
||||||
labeled_df = JoinLabelsStep(context).transform(features_df)
|
|
||||||
|
|
||||||
return labeled_df
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == '__main__':
|
if __name__ == '__main__':
|
||||||
|
|
||||||
class ExperimentsProvider(SupabaseProvider, BackendAPIProvider):
|
class Provider(SupabaseProvider, BackendAPIProvider):
|
||||||
|
def __init__(self, backend_url: str):
|
||||||
|
SupabaseProvider.__init__(self)
|
||||||
|
BackendAPIProvider.__init__(self, backend_url=backend_url)
|
||||||
|
|
||||||
|
|
||||||
|
class HistoricalProvider(SupabaseProvider, BackendAPIProvider):
|
||||||
def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
|
def fetch_kafka_topic(self, topic: str) -> pd.DataFrame:
|
||||||
base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
|
path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
|
||||||
if not os.path.isdir(base_path):
|
interactions_file = "messages(2).json"
|
||||||
return pd.DataFrame()
|
prices_file = "messages(3).json"
|
||||||
|
|
||||||
files = {"user-interactions": "int.json", "price-logs": "price.json"}
|
data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
|
||||||
file_to_read = files.get(topic, files["user-interactions"])
|
data = [r['payload'] for r in data['value'].to_list()]
|
||||||
frames = []
|
data = pd.DataFrame(data)
|
||||||
|
return data
|
||||||
|
|
||||||
for d in os.listdir(base_path):
|
|
||||||
full_path = os.path.join(base_path, d, file_to_read)
|
|
||||||
if not os.path.isfile(full_path):
|
|
||||||
continue
|
|
||||||
try:
|
|
||||||
data = pd.read_json(full_path)
|
|
||||||
payloads = pd.DataFrame([r['payload'] for r in data['value'].to_list()])
|
|
||||||
frames.append(payloads)
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Warning: Could not process {full_path}: {e}")
|
|
||||||
|
|
||||||
return pd.concat(frames, ignore_index=True) if frames else pd.DataFrame()
|
# example run
|
||||||
|
context = PipelineContext(
|
||||||
|
provider=HistoricalProvider(),
|
||||||
|
store_mode='hotel',
|
||||||
|
)
|
||||||
|
|
||||||
# demo: run ML training pipeline
|
product_features, prices = full_pipeline(context)
|
||||||
context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
|
print(prices.to_string())
|
||||||
features = ml_training_pipeline(context)
|
|
||||||
print(f"Feature matrix: {features.shape}")
|
|
||||||
print(features.head())
|
|
||||||
print(features.info())
|
|
||||||
|
|
||||||
features.to_parquet("features.parquet")
|
|
||||||
|
|||||||
@@ -18,17 +18,10 @@ class SupabaseProvider(DataProvider):
|
|||||||
self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
|
self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
|
||||||
|
|
||||||
def fetch_products(self, store_mode: str) -> pd.DataFrame:
|
def fetch_products(self, store_mode: str) -> pd.DataFrame:
|
||||||
# hotel uses room_type, airline uses flight_type; select all and normalize
|
resp = self.supabase.table(f'{store_mode}_products').select(
|
||||||
resp = self.supabase.table(f'{store_mode}_products').select("*").execute()
|
"id, room_type, date_index, metadata, availability"
|
||||||
if not resp.data:
|
).execute()
|
||||||
return pd.DataFrame()
|
return pd.DataFrame(resp.data) if resp.data else pd.DataFrame()
|
||||||
df = pd.DataFrame(resp.data)
|
|
||||||
# normalize type column: hotel has room_type, airline has flight_type
|
|
||||||
if 'room_type' in df.columns:
|
|
||||||
df['product_type'] = df['room_type']
|
|
||||||
elif 'flight_type' in df.columns:
|
|
||||||
df['product_type'] = df['flight_type']
|
|
||||||
return df
|
|
||||||
|
|
||||||
def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
|
def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
|
||||||
if not experiment_ids:
|
if not experiment_ids:
|
||||||
|
|||||||
@@ -6,11 +6,7 @@ from procesing.steps.chunk import ChunkByTimeWindowStep
|
|||||||
from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
|
from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
|
||||||
from procesing.steps.elasticity import AggregatePriceLogsStep
|
from procesing.steps.elasticity import AggregatePriceLogsStep
|
||||||
from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
|
from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
|
||||||
from procesing.steps.session import (
|
from procesing.steps.session import ExtractSessionFeaturesStep, _extract_features_for_session
|
||||||
ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
|
|
||||||
TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
|
|
||||||
_extract_features_for_session
|
|
||||||
)
|
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
'BaseContextStep',
|
'BaseContextStep',
|
||||||
@@ -29,11 +25,5 @@ __all__ = [
|
|||||||
'FitPricingFunctionStep',
|
'FitPricingFunctionStep',
|
||||||
'PredictPricesStep',
|
'PredictPricesStep',
|
||||||
'ExtractSessionFeaturesStep',
|
'ExtractSessionFeaturesStep',
|
||||||
'JoinLabelsStep',
|
|
||||||
'ValidateDataStep',
|
|
||||||
'TemporalFeatureStep',
|
|
||||||
'BehavioralFeatureStep',
|
|
||||||
'ProductFeatureStep',
|
|
||||||
'UserAgentFeatureStep',
|
|
||||||
'_extract_features_for_session',
|
'_extract_features_for_session',
|
||||||
]
|
]
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
from sklearn.base import BaseEstimator, TransformerMixin
|
from sklearn.base import BaseEstimator, TransformerMixin
|
||||||
from procesing.context import PipelineContext
|
from procesing.context import PipelineContext
|
||||||
from typing import Any
|
|
||||||
|
|
||||||
class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
|
class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
|
||||||
"""
|
"""
|
||||||
@@ -17,7 +16,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
|
|||||||
return self
|
return self
|
||||||
|
|
||||||
@abstractmethod
|
@abstractmethod
|
||||||
def transform(self, X) -> Any:
|
def transform(self, X):
|
||||||
"""Transform input using context. Must be implemented by subclass."""
|
"""Transform input using context. Must be implemented by subclass."""
|
||||||
pass
|
pass
|
||||||
|
|
||||||
|
|||||||
@@ -7,12 +7,12 @@ class AggregatePriceLogsStep(BaseContextStep):
|
|||||||
"""
|
"""
|
||||||
Aggregate price logs into time windows using VECTORIZED operations.
|
Aggregate price logs into time windows using VECTORIZED operations.
|
||||||
Input: price_logs_df
|
Input: price_logs_df
|
||||||
Output: DataFrame with columns [productId, price]
|
Output: list of price chunks with [productId, price]
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def transform(self, price_logs_df: pd.DataFrame):
|
def transform(self, price_logs_df: pd.DataFrame):
|
||||||
if price_logs_df.empty:
|
if price_logs_df.empty:
|
||||||
return pd.DataFrame(columns=['productId', 'price'])
|
return []
|
||||||
|
|
||||||
df = price_logs_df.copy()
|
df = price_logs_df.copy()
|
||||||
ts_col = self.context.config.get('ts_col', 'ts')
|
ts_col = self.context.config.get('ts_col', 'ts')
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import pandas as pd
|
|||||||
from procesing.steps.base import BaseContextStep
|
from procesing.steps.base import BaseContextStep
|
||||||
|
|
||||||
class FetchInteractionsStep(BaseContextStep):
|
class FetchInteractionsStep(BaseContextStep):
|
||||||
"""Fetch raw interaction data from Kafka topic with optional time and store_mode filtering"""
|
"""Fetch raw interaction data from Kafka topic with optional time filtering"""
|
||||||
|
|
||||||
def __init__(self, context, lookback: str = None):
|
def __init__(self, context, lookback: str = None):
|
||||||
super().__init__(context)
|
super().__init__(context)
|
||||||
@@ -24,10 +24,6 @@ class FetchInteractionsStep(BaseContextStep):
|
|||||||
# drop all where page has /admin/
|
# drop all where page has /admin/
|
||||||
df = df[~df['page'].str.contains('/admin/', na=False)]
|
df = df[~df['page'].str.contains('/admin/', na=False)]
|
||||||
|
|
||||||
# filter by store_mode from context
|
|
||||||
if 'storeMode' in df.columns:
|
|
||||||
df = df[df['storeMode'] == self.context.store_mode]
|
|
||||||
|
|
||||||
# Remap dateIndex if present
|
# Remap dateIndex if present
|
||||||
if 'metadata_dateIndex' in df.columns:
|
if 'metadata_dateIndex' in df.columns:
|
||||||
df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
|
df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
|
||||||
@@ -42,7 +38,7 @@ class FetchInteractionsStep(BaseContextStep):
|
|||||||
|
|
||||||
|
|
||||||
class FetchPriceLogsStep(BaseContextStep):
|
class FetchPriceLogsStep(BaseContextStep):
|
||||||
"""Fetch price log data from Kafka topic with optional time and store_mode filtering"""
|
"""Fetch price log data from Kafka topic with optional time filtering"""
|
||||||
|
|
||||||
def __init__(self, context, lookback: str = None):
|
def __init__(self, context, lookback: str = None):
|
||||||
super().__init__(context)
|
super().__init__(context)
|
||||||
@@ -54,10 +50,6 @@ class FetchPriceLogsStep(BaseContextStep):
|
|||||||
if df.empty:
|
if df.empty:
|
||||||
return df
|
return df
|
||||||
|
|
||||||
# filter by store_mode from context
|
|
||||||
if 'storeMode' in df.columns:
|
|
||||||
df = df[df['storeMode'] == self.context.store_mode]
|
|
||||||
|
|
||||||
# Apply time filtering if lookback specified
|
# Apply time filtering if lookback specified
|
||||||
if self.lookback and 'ts' in df.columns:
|
if self.lookback and 'ts' in df.columns:
|
||||||
df['ts'] = pd.to_datetime(df['ts'])
|
df['ts'] = pd.to_datetime(df['ts'])
|
||||||
|
|||||||
@@ -1,261 +1,159 @@
|
|||||||
"""
|
"""
|
||||||
Session feature extraction for ML training pipeline.
|
Session feature extraction for S_t component of state space.
|
||||||
|
Computes behavioral signals from interaction data already in pipeline.
|
||||||
"""
|
"""
|
||||||
import pandas as pd
|
import pandas as pd
|
||||||
import numpy as np
|
import numpy as np
|
||||||
import re
|
from typing import Optional, Dict, Any
|
||||||
from typing import Dict, Any
|
from collections import Counter
|
||||||
from procesing.steps.base import BaseContextStep
|
from procesing.steps.base import BaseContextStep
|
||||||
|
|
||||||
EVENT_CATS = {
|
def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
|
||||||
'page_view': ['page_view'],
|
"""Compute features for single session.
|
||||||
'item_view': ['view_item_page', 'learn_more_about_item'],
|
|
||||||
'cart_add': ['add_item_to_cart'],
|
Args:
|
||||||
'purchase': ['purchase', 'checkout_complete'],
|
session_df: interaction events for this session
|
||||||
'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
|
session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
|
||||||
# 'filter': ['filter', 'search', 'apply_filter'],
|
"""
|
||||||
}
|
features = {}
|
||||||
HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
|
|
||||||
AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
|
# basic counts
|
||||||
BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
|
features['total_interactions'] = len(session_df)
|
||||||
('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
|
|
||||||
|
event_counts = session_df['eventName'].value_counts().to_dict()
|
||||||
|
features['page_views'] = event_counts.get('page_view', 0) + event_counts.get('view_item_page', 0)
|
||||||
|
features['item_views'] = event_counts.get('view_item_page', 0)
|
||||||
|
features['searches'] = event_counts.get('search', 0)
|
||||||
|
features['cart_adds'] = event_counts.get('add_item_to_cart', 0)
|
||||||
|
|
||||||
|
# hover events
|
||||||
|
hover_events = ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button']
|
||||||
|
features['hovers'] = sum(event_counts.get(ev, 0) for ev in hover_events)
|
||||||
|
|
||||||
|
# product-level signals
|
||||||
|
product_ids = session_df['productId'].dropna()
|
||||||
|
features['unique_products_viewed'] = product_ids.nunique()
|
||||||
|
|
||||||
|
if len(product_ids) > 0:
|
||||||
|
product_view_counts = Counter(product_ids)
|
||||||
|
features['product_view_depth'] = max(product_view_counts.values())
|
||||||
|
else:
|
||||||
|
features['product_view_depth'] = 0
|
||||||
|
|
||||||
|
# temporal features with session timeout logic
|
||||||
|
if 'ts' in session_df.columns:
|
||||||
|
timestamps = session_df['ts'].sort_values()
|
||||||
|
|
||||||
|
# compute active duration considering timeout gaps
|
||||||
|
if len(timestamps) > 1:
|
||||||
|
time_diffs = timestamps.diff().dropna().dt.total_seconds()
|
||||||
|
# only count gaps shorter than timeout towards active session duration
|
||||||
|
active_diffs = time_diffs[time_diffs <= session_timeout_sec]
|
||||||
|
features['session_duration_sec'] = active_diffs.sum() if len(active_diffs) > 0 else 0.0
|
||||||
|
|
||||||
|
features['avg_time_between_events'] = time_diffs.mean()
|
||||||
|
features['std_time_between_events'] = time_diffs.std()
|
||||||
|
else:
|
||||||
|
features['session_duration_sec'] = 0.0
|
||||||
|
features['avg_time_between_events'] = 0.0
|
||||||
|
features['std_time_between_events'] = 0.0
|
||||||
|
|
||||||
|
if features['session_duration_sec'] > 0:
|
||||||
|
features['interaction_velocity'] = (features['total_interactions'] / features['session_duration_sec']) * 60
|
||||||
|
else:
|
||||||
|
features['interaction_velocity'] = 0.0
|
||||||
|
else:
|
||||||
|
features['session_duration_sec'] = 0.0
|
||||||
|
features['interaction_velocity'] = 0.0
|
||||||
|
features['avg_time_between_events'] = 0.0
|
||||||
|
features['std_time_between_events'] = 0.0
|
||||||
|
|
||||||
|
# cart/conversion signals
|
||||||
|
features['cart_to_view_ratio'] = features['cart_adds'] / features['item_views'] if features['item_views'] > 0 else 0.0
|
||||||
|
|
||||||
|
return features
|
||||||
|
|
||||||
|
|
||||||
def _get_browser(s: str) -> str:
|
def _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
|
||||||
if pd.isna(s): return 'Unknown'
|
"""Apply feature extraction to sliding window of interactions."""
|
||||||
for name, pat in BROWSER_PATTERNS:
|
# add columns of all features at each step
|
||||||
if re.search(pat, s): return name
|
new_cols = ["total_interactions", "page_views", "item_views", "searches",
|
||||||
return 'Other'
|
"cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
|
||||||
|
"session_duration_sec", "interaction_velocity",
|
||||||
|
"avg_time_between_events", "std_time_between_events",
|
||||||
|
"cart_to_view_ratio"]
|
||||||
|
for col in new_cols: df[col] = np.nan
|
||||||
|
for idx in range(1, len(df) + 1):
|
||||||
|
features = _extract_features_for_session(df.iloc[:idx])
|
||||||
|
# fillna kinda meh
|
||||||
|
features = { k: (v if not pd.isna(v) else 0.0) for k, v in features.items() }
|
||||||
|
for col in new_cols:
|
||||||
|
df.at[df.index[idx - 1], col] = features[col]
|
||||||
|
#print(f"Processed {idx}/{len(df)} events for session {df['sessionId'].iloc[0]}")
|
||||||
|
return df
|
||||||
|
|
||||||
|
class BuildStateSpaceStep(BaseContextStep):
|
||||||
|
"""
|
||||||
|
Build state space representation S_t from session features.
|
||||||
|
|
||||||
|
Input: session_features DataFrame
|
||||||
|
Output: state_space_df DataFrame with S_t vectors
|
||||||
|
"""
|
||||||
|
|
||||||
|
def transform(self, rich_dataset: pd.DataFrame) -> pd.DataFrame:
|
||||||
|
# check if features are present
|
||||||
|
required_cols = ["total_interactions", "page_views", "item_views", "searches",
|
||||||
|
"cart_adds", "hovers", "unique_products_viewed", "product_view_depth",
|
||||||
|
"session_duration_sec", "interaction_velocity",
|
||||||
|
"avg_time_between_events", "std_time_between_events",
|
||||||
|
"cart_to_view_ratio"]
|
||||||
|
if not all(col in rich_dataset.columns for col in required_cols):
|
||||||
|
raise ValueError("Missing required columns for feature extraction.")
|
||||||
|
if rich_dataset.empty:
|
||||||
|
return pd.DataFrame()
|
||||||
|
|
||||||
|
|
||||||
class TemporalFeatureStep(BaseContextStep):
|
# For simplicity, we return as is
|
||||||
"""Vectorized time-based features: durations, velocities, gaps."""
|
return rich_dataset.copy()
|
||||||
|
|
||||||
def __init__(self, context, timeout_sec: float = 900, velocity_window: str = '5min'):
|
|
||||||
super().__init__(context)
|
|
||||||
self.timeout_sec = timeout_sec
|
|
||||||
self.velocity_window = velocity_window
|
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
|
|
||||||
df = X.copy()
|
|
||||||
if df.empty or 'ts' not in df.columns:
|
|
||||||
return pd.DataFrame(columns=pd.Series(['sessionId']))
|
|
||||||
|
|
||||||
df['ts_dt'] = pd.to_datetime(df['ts'])
|
|
||||||
df = df.sort_values(['sessionId', 'ts_dt'])
|
|
||||||
df['time_diff'] = df.groupby('sessionId')['ts_dt'].diff().dt.total_seconds()
|
|
||||||
df['active_diff'] = df['time_diff'].where(df['time_diff'] <= self.timeout_sec, 0)
|
|
||||||
|
|
||||||
agg = df.groupby('sessionId').agg(
|
|
||||||
session_duration_sec=('active_diff', 'sum'),
|
|
||||||
total_interactions=('sessionId', 'count'),
|
|
||||||
avg_time_between_events=('time_diff', 'mean'),
|
|
||||||
std_time_between_events=('time_diff', 'std'),
|
|
||||||
min_time_between_events=('time_diff', 'min'),
|
|
||||||
session_start_hour=('ts_dt', lambda x: x.min().hour),
|
|
||||||
).reset_index()
|
|
||||||
agg['std_time_between_events'] = agg['std_time_between_events'].fillna(0)
|
|
||||||
agg['interaction_velocity'] = np.where(
|
|
||||||
agg['session_duration_sec'] > 0,
|
|
||||||
(agg['total_interactions'] / agg['session_duration_sec']) * 60, 0)
|
|
||||||
|
|
||||||
vel = df.set_index('ts_dt').groupby('sessionId').resample(self.velocity_window, include_groups=False).size()
|
|
||||||
max_velocity = vel.groupby('sessionId').max().rename('max_velocity_5min')
|
|
||||||
agg = agg.merge(max_velocity, on='sessionId', how='left')
|
|
||||||
agg['max_velocity_5min'] = agg['max_velocity_5min'].fillna(0)
|
|
||||||
return agg
|
|
||||||
|
|
||||||
|
|
||||||
class BehavioralFeatureStep(BaseContextStep):
|
|
||||||
"""Vectorized event counts and ratios per session."""
|
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
|
|
||||||
df = X.copy()
|
|
||||||
if df.empty or 'eventName' not in df.columns:
|
|
||||||
return pd.DataFrame(columns=pd.Series(['sessionId']))
|
|
||||||
|
|
||||||
for cat, events in EVENT_CATS.items():
|
|
||||||
df[f'is_{cat}'] = df['eventName'].isin(events)
|
|
||||||
df['is_hover'] = df['is_hover'] | df['eventName'].str.startswith('hover_over_')
|
|
||||||
|
|
||||||
agg = df.groupby('sessionId').agg(
|
|
||||||
total_events=('eventName', 'count'), unique_pages=('page', 'nunique'),
|
|
||||||
page_views=('is_page_view', 'sum'), item_views=('is_item_view', 'sum'),
|
|
||||||
cart_adds=('is_cart_add', 'sum'), purchases=('is_purchase', 'sum'),
|
|
||||||
hover_events=('is_hover', 'sum'),
|
|
||||||
# filter_events=('is_filter', 'sum'),
|
|
||||||
).reset_index()
|
|
||||||
agg['cart_to_view_ratio'] = np.where(agg['item_views'] > 0, agg['cart_adds'] / agg['item_views'], 0)
|
|
||||||
agg['conversion_rate'] = np.where(agg['item_views'] > 0, agg['purchases'] / agg['item_views'], 0)
|
|
||||||
agg['hover_intensity'] = np.where(agg['total_events'] > 0, agg['hover_events'] / agg['total_events'], 0)
|
|
||||||
return agg
|
|
||||||
|
|
||||||
|
|
||||||
class ProductFeatureStep(BaseContextStep):
|
|
||||||
"""Vectorized product interaction features: diversity, depth, price sensitivity."""
|
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
|
|
||||||
df = X.copy()
|
|
||||||
if df.empty:
|
|
||||||
return pd.DataFrame(columns=pd.Series(['sessionId']))
|
|
||||||
price_col = next((c for c in ['metadata_base_price', 'metadata_price', 'base_price'] if c in df.columns), None)
|
|
||||||
df['price_seen'] = pd.to_numeric(df[price_col], errors='coerce') if price_col else np.nan
|
|
||||||
|
|
||||||
prod_df = df[df['productId'].notna()]
|
|
||||||
if prod_df.empty:
|
|
||||||
return pd.DataFrame(columns=pd.Series(['sessionId', 'unique_products_viewed', 'product_view_depth', 'avg_price_seen', 'min_price_seen', 'max_price_seen', 'price_range']))
|
|
||||||
|
|
||||||
agg = prod_df.groupby('sessionId').agg(
|
|
||||||
unique_products_viewed=('productId', 'nunique'),
|
|
||||||
product_view_depth=('productId', lambda x: x.value_counts().iloc[0] if len(x) > 0 else 0),
|
|
||||||
avg_price_seen=('price_seen', 'mean'), min_price_seen=('price_seen', 'min'),
|
|
||||||
max_price_seen=('price_seen', 'max'),
|
|
||||||
).reset_index()
|
|
||||||
agg['price_range'] = (agg['max_price_seen'] - agg['min_price_seen']).fillna(0)
|
|
||||||
return agg
|
|
||||||
|
|
||||||
|
|
||||||
class UserAgentFeatureStep(BaseContextStep):
|
|
||||||
"""Parse userAgent into bot-detection signals."""
|
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame|pd.Series:
|
|
||||||
df = X.copy()
|
|
||||||
if df.empty or 'userAgent' not in df.columns:
|
|
||||||
return pd.DataFrame(columns=pd.Series(['sessionId']))
|
|
||||||
|
|
||||||
ua = df.groupby('sessionId')['userAgent'].first().reset_index()
|
|
||||||
ua['is_headless'] = ua['userAgent'].str.contains(HEADLESS_RE, na=False)
|
|
||||||
ua['is_automation'] = ua['userAgent'].str.contains(AUTOMATION_RE, na=False)
|
|
||||||
ua['browser_family'] = ua['userAgent'].apply(_get_browser)
|
|
||||||
return ua[['sessionId', 'is_headless', 'is_automation', 'browser_family']]
|
|
||||||
|
|
||||||
|
|
||||||
class ExtractSessionFeaturesStep(BaseContextStep):
|
class ExtractSessionFeaturesStep(BaseContextStep):
|
||||||
"""
|
"""
|
||||||
Vectorized session feature extraction - replaces O(n^2) per-row loop.
|
Extract session-level behavioral features from interaction logs.
|
||||||
Input: interactions_df
|
|
||||||
Output: session-level feature matrix
|
Input: interactions_df (user-interactions from earlier pipeline step)
|
||||||
|
Output: interactions_df with added session feature columns
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
|
def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
|
||||||
if X.empty:
|
if interactions_df.empty:
|
||||||
return pd.DataFrame()
|
return pd.DataFrame()
|
||||||
df = X.copy()
|
|
||||||
|
|
||||||
# run all feature steps and merge on sessionId
|
# ensure timestamp column
|
||||||
temporal = TemporalFeatureStep(self.context).transform(df)
|
if 'ts' in interactions_df.columns:
|
||||||
behavioral = BehavioralFeatureStep(self.context).transform(df)
|
interactions_df = interactions_df.copy()
|
||||||
product = ProductFeatureStep(self.context).transform(df)
|
interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])
|
||||||
ua = UserAgentFeatureStep(self.context).transform(df)
|
|
||||||
|
|
||||||
result = temporal
|
# group by session and compute features
|
||||||
for other in [behavioral, product, ua]:
|
session_features = []
|
||||||
if not other.empty and 'sessionId' in other.columns:
|
for session_id, session_df in interactions_df.groupby('sessionId'):
|
||||||
result = result.merge(other, on='sessionId', how='left')
|
new_slice = _apply_to_slice(session_df.sort_values('ts'))
|
||||||
|
session_features.append(new_slice)
|
||||||
|
|
||||||
# carry forward experimentId for label joining
|
return pd.concat(session_features, ignore_index=True)
|
||||||
if 'experimentId' in df.columns:
|
|
||||||
exp_map = df.groupby('sessionId')['experimentId'].first()
|
|
||||||
result = result.merge(exp_map, on='sessionId', how='left')
|
|
||||||
|
|
||||||
return result
|
|
||||||
|
|
||||||
|
|
||||||
class JoinLabelsStep(BaseContextStep):
|
|
||||||
|
class FilterSessionInteractionsStep(BaseContextStep):
|
||||||
"""
|
"""
|
||||||
Join experiment labels to session features.
|
Filter interactions DataFrame to specific session.
|
||||||
Input: (features_df, experiments_df) or features_df (fetches experiments)
|
|
||||||
Output: labeled feature matrix with is_agent column
|
Input: (interactions_df, session_id)
|
||||||
|
Output: interactions_df filtered to session_id
|
||||||
"""
|
"""
|
||||||
|
|
||||||
def transform(self, X : tuple) -> pd.DataFrame:
|
def transform(self, data: tuple) -> pd.DataFrame:
|
||||||
data = X;
|
interactions_df, session_id = data
|
||||||
if isinstance(data, tuple):
|
return interactions_df[interactions_df['sessionId'] == session_id].copy()
|
||||||
features_df, experiments_df = data
|
|
||||||
else:
|
|
||||||
features_df = data
|
|
||||||
if 'experimentId' not in features_df.columns:
|
|
||||||
return features_df
|
|
||||||
exp_ids = features_df['experimentId'].dropna().unique().tolist()
|
|
||||||
experiments_df = self.context.provider.fetch_experiments(exp_ids) if exp_ids else pd.DataFrame()
|
|
||||||
|
|
||||||
if features_df.empty:
|
|
||||||
return features_df
|
|
||||||
if experiments_df.empty:
|
|
||||||
features_df['is_agent'] = np.nan
|
|
||||||
return features_df
|
|
||||||
|
|
||||||
exp = experiments_df.copy()
|
|
||||||
if 'id' in exp.columns:
|
|
||||||
exp = exp.rename(columns={'id': 'experimentId'})
|
|
||||||
if 'xp_human_only' in exp.columns:
|
|
||||||
exp['is_agent'] = ~exp['xp_human_only']
|
|
||||||
|
|
||||||
cols = ['experimentId'] + [c for c in ['is_agent', 'xp_human_only', 'xp_market_mode'] if c in exp.columns]
|
|
||||||
return features_df.merge(exp[cols].drop_duplicates(), on='experimentId', how='left')
|
|
||||||
|
|
||||||
|
|
||||||
class ValidateDataStep(BaseContextStep):
|
|
||||||
"""
|
|
||||||
Data quality checks before training.
|
|
||||||
Input: df
|
|
||||||
Output: df (unchanged, but logs validation report to context)
|
|
||||||
"""
|
|
||||||
REQUIRED = ['sessionId', 'eventName', 'ts']
|
|
||||||
|
|
||||||
def transform(self, X: pd.DataFrame) -> pd.DataFrame:
|
|
||||||
df = X.copy()
|
|
||||||
report = {'status': 'valid', 'rows': len(df), 'sessions': 0}
|
|
||||||
if df.empty:
|
|
||||||
report['status'] = 'empty'
|
|
||||||
self.context.cache('validation_report', report)
|
|
||||||
return df
|
|
||||||
|
|
||||||
missing = [c for c in self.REQUIRED if c not in df.columns]
|
|
||||||
if missing:
|
|
||||||
report['status'] = 'invalid'
|
|
||||||
report['missing_cols'] = missing
|
|
||||||
|
|
||||||
report['sessions'] = df['sessionId'].nunique() if 'sessionId' in df.columns else 0
|
|
||||||
report['null_sessions'] = int(df['sessionId'].isna().sum()) if 'sessionId' in df.columns else 0
|
|
||||||
if 'experimentId' in df.columns:
|
|
||||||
report['null_experiments'] = int(df['experimentId'].isna().sum())
|
|
||||||
|
|
||||||
self.context.cache('validation_report', report)
|
|
||||||
return df
|
|
||||||
|
|
||||||
|
|
||||||
# legacy compat - kept for backwards compatibility with existing code
|
|
||||||
def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
|
|
||||||
"""Single-session feature extraction (legacy interface)."""
|
|
||||||
defaults = {k: 0 for k in ['total_interactions', 'page_views', 'item_views', 'searches',
|
|
||||||
'cart_adds', 'hovers', 'unique_products_viewed', 'product_view_depth',
|
|
||||||
'session_duration_sec', 'interaction_velocity',
|
|
||||||
'avg_time_between_events', 'std_time_between_events', 'cart_to_view_ratio']}
|
|
||||||
if session_df.empty:
|
|
||||||
return defaults
|
|
||||||
|
|
||||||
session_df = session_df.copy()
|
|
||||||
if 'sessionId' not in session_df.columns:
|
|
||||||
session_df['sessionId'] = 'tmp'
|
|
||||||
|
|
||||||
# use a dummy context for the steps
|
|
||||||
class DummyCtx: config = {} # should maybe inherit but whatever
|
|
||||||
ctx = DummyCtx()
|
|
||||||
|
|
||||||
t = TemporalFeatureStep(ctx, timeout_sec=session_timeout_sec).transform(session_df)
|
|
||||||
b = BehavioralFeatureStep(ctx).transform(session_df)
|
|
||||||
p = ProductFeatureStep(ctx).transform(session_df)
|
|
||||||
|
|
||||||
result = {}
|
|
||||||
for df in [t, b, p]:
|
|
||||||
if not df.empty:
|
|
||||||
for col in df.columns:
|
|
||||||
if col != 'sessionId':
|
|
||||||
result[col] = df[col].iloc[0] if len(df) > 0 else 0
|
|
||||||
|
|
||||||
remap = {'hover_events': 'hovers', 'filter_events': 'searches', 'unique_pages': 'unique_pages_visited'}
|
|
||||||
for old, new in remap.items():
|
|
||||||
if old in result:
|
|
||||||
result[new] = result.pop(old)
|
|
||||||
return result
|
|
||||||
|
|||||||
@@ -144,7 +144,7 @@ def mock_price_logs_raw_kafka():
|
|||||||
'price': 162.47,
|
'price': 162.47,
|
||||||
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
||||||
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
||||||
'storeMode': 'hotel',
|
'storeMode': 'shop',
|
||||||
'ts': '2025-11-25T21:05:57.967Z'
|
'ts': '2025-11-25T21:05:57.967Z'
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -157,7 +157,7 @@ def mock_price_logs_raw_kafka():
|
|||||||
'price': 743.49,
|
'price': 743.49,
|
||||||
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
||||||
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
||||||
'storeMode': 'hotel',
|
'storeMode': 'shop',
|
||||||
'ts': '2025-11-25T21:05:57.993Z'
|
'ts': '2025-11-25T21:05:57.993Z'
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -170,7 +170,7 @@ def mock_price_logs_raw_kafka():
|
|||||||
'price': 163.87,
|
'price': 163.87,
|
||||||
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
||||||
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
||||||
'storeMode': 'hotel',
|
'storeMode': 'shop',
|
||||||
'ts': '2025-11-25T21:05:58.009Z'
|
'ts': '2025-11-25T21:05:58.009Z'
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -183,7 +183,7 @@ def mock_price_logs_raw_kafka():
|
|||||||
'price': 397.46,
|
'price': 397.46,
|
||||||
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
||||||
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
||||||
'storeMode': 'hotel',
|
'storeMode': 'shop',
|
||||||
'ts': '2025-11-25T21:05:58.049Z'
|
'ts': '2025-11-25T21:05:58.049Z'
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -196,7 +196,7 @@ def mock_price_logs_raw_kafka():
|
|||||||
'price': 401.66,
|
'price': 401.66,
|
||||||
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
|
||||||
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
|
||||||
'storeMode': 'hotel',
|
'storeMode': 'shop',
|
||||||
'ts': '2025-11-25T21:06:08.864Z'
|
'ts': '2025-11-25T21:06:08.864Z'
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -222,7 +222,7 @@ def mock_experiments():
|
|||||||
'created_at': pd.to_datetime(['2025-11-25T20:00:00Z', '2025-11-26T10:00:00Z']),
|
'created_at': pd.to_datetime(['2025-11-25T20:00:00Z', '2025-11-26T10:00:00Z']),
|
||||||
'subject_name': ['Session A', 'Session B'],
|
'subject_name': ['Session A', 'Session B'],
|
||||||
'xp_human_only': [True, False],
|
'xp_human_only': [True, False],
|
||||||
'xp_market_mode': ['hotel', 'airline'],
|
'xp_market_mode': ['hotel', 'shop'],
|
||||||
'xp_task_id': [None, None]
|
'xp_task_id': [None, None]
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -269,13 +269,3 @@ def empty_context(empty_provider):
|
|||||||
store_mode='hotel',
|
store_mode='hotel',
|
||||||
window_size='30s'
|
window_size='30s'
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def session_interactions(mock_interactions):
|
|
||||||
"""Enriched interaction data for session feature extraction tests"""
|
|
||||||
df = mock_interactions.copy()
|
|
||||||
df['userAgent'] = ['Mozilla/5.0 Chrome/120', 'Mozilla/5.0 Chrome/120',
|
|
||||||
'HeadlessChrome/120', 'HeadlessChrome/120', 'HeadlessChrome/120']
|
|
||||||
df['metadata_base_price'] = [None, None, 150.0, 150.0, 200.0]
|
|
||||||
return df
|
|
||||||
|
|||||||
@@ -2,20 +2,10 @@
|
|||||||
|
|
||||||
import { useState, FormEvent } from 'react';
|
import { useState, FormEvent } from 'react';
|
||||||
import { useRouter } from 'next/navigation';
|
import { useRouter } from 'next/navigation';
|
||||||
import { Button, Label, DateInput, Dropdown, DropdownCounter, SelectDropdown, SelectOption } from '@/components/ui';
|
import { Button, Label, Input, DateInput, RadioGroup, Dropdown, DropdownCounter } from '@/components/ui';
|
||||||
import { dateToDaysFromToday } from '@/lib/airline-utils';
|
import { dateToDaysFromToday } from '@/lib/airline-utils';
|
||||||
|
|
||||||
const CITIES: SelectOption[] = [
|
type TripType = 'roundtrip' | 'oneway' | 'multicity';
|
||||||
{ value: 'JFK', label: 'New York (JFK)', sublabel: 'John F. Kennedy International' },
|
|
||||||
{ value: 'LAX', label: 'Los Angeles (LAX)', sublabel: 'Los Angeles International' },
|
|
||||||
{ value: 'ORD', label: 'Chicago (ORD)', sublabel: "O'Hare International" },
|
|
||||||
{ value: 'MIA', label: 'Miami (MIA)', sublabel: 'Miami International' },
|
|
||||||
{ value: 'SFO', label: 'San Francisco (SFO)', sublabel: 'San Francisco International' },
|
|
||||||
{ value: 'SEA', label: 'Seattle (SEA)', sublabel: 'Seattle-Tacoma International' },
|
|
||||||
{ value: 'ATL', label: 'Atlanta (ATL)', sublabel: 'Hartsfield-Jackson International' },
|
|
||||||
{ value: 'DFW', label: 'Dallas (DFW)', sublabel: 'Dallas/Fort Worth International' },
|
|
||||||
];
|
|
||||||
|
|
||||||
|
|
||||||
const PlaneIcon = () => (
|
const PlaneIcon = () => (
|
||||||
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
|
<svg className="w-5 h-5" fill="none" stroke="currentColor" viewBox="0 0 24 24">
|
||||||
@@ -32,9 +22,11 @@ const LocationIcon = () => (
|
|||||||
|
|
||||||
export default function AirlineHero() {
|
export default function AirlineHero() {
|
||||||
const router = useRouter();
|
const router = useRouter();
|
||||||
|
const [tripType, setTripType] = useState<TripType>('roundtrip');
|
||||||
const [origin, setOrigin] = useState('');
|
const [origin, setOrigin] = useState('');
|
||||||
const [destination, setDestination] = useState('');
|
const [destination, setDestination] = useState('');
|
||||||
const [departDate, setDepartDate] = useState('');
|
const [departDate, setDepartDate] = useState('');
|
||||||
|
const [returnDate, setReturnDate] = useState('');
|
||||||
const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });
|
const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });
|
||||||
|
|
||||||
const handleSearch = (e: FormEvent) => {
|
const handleSearch = (e: FormEvent) => {
|
||||||
@@ -48,6 +40,8 @@ export default function AirlineHero() {
|
|||||||
|
|
||||||
if (origin) params.set('origin', origin);
|
if (origin) params.set('origin', origin);
|
||||||
if (destination) params.set('destination', destination);
|
if (destination) params.set('destination', destination);
|
||||||
|
if (tripType !== 'roundtrip') params.set('tripType', tripType);
|
||||||
|
if (returnDate && tripType === 'roundtrip') params.set('returnDate', returnDate);
|
||||||
|
|
||||||
params.set('adults', passengers.adults.toString());
|
params.set('adults', passengers.adults.toString());
|
||||||
params.set('children', passengers.children.toString());
|
params.set('children', passengers.children.toString());
|
||||||
@@ -72,15 +66,28 @@ export default function AirlineHero() {
|
|||||||
|
|
||||||
<div className="search-form">
|
<div className="search-form">
|
||||||
<form onSubmit={handleSearch}>
|
<form onSubmit={handleSearch}>
|
||||||
<div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-3 gap-4">
|
<div className="mb-6">
|
||||||
|
<RadioGroup
|
||||||
|
name="tripType"
|
||||||
|
value={tripType}
|
||||||
|
onChange={setTripType}
|
||||||
|
options={[
|
||||||
|
{ value: 'roundtrip', label: 'Round-trip' },
|
||||||
|
{ value: 'oneway', label: 'One-way' },
|
||||||
|
{ value: 'multicity', label: 'Multi-city' },
|
||||||
|
]}
|
||||||
|
/>
|
||||||
|
</div>
|
||||||
|
|
||||||
|
<div className="grid grid-cols-1 sm:grid-cols-2 lg:grid-cols-4 gap-4">
|
||||||
<div>
|
<div>
|
||||||
<Label htmlFor="origin">From</Label>
|
<Label htmlFor="origin">From</Label>
|
||||||
<SelectDropdown
|
<Input
|
||||||
|
type="text"
|
||||||
id="origin"
|
id="origin"
|
||||||
value={origin}
|
value={origin}
|
||||||
onChange={setOrigin}
|
onChange={(e) => setOrigin(e.target.value)}
|
||||||
options={CITIES}
|
placeholder="Airport or city"
|
||||||
placeholder="Select origin"
|
|
||||||
icon={<PlaneIcon />}
|
icon={<PlaneIcon />}
|
||||||
required
|
required
|
||||||
/>
|
/>
|
||||||
@@ -88,12 +95,12 @@ export default function AirlineHero() {
|
|||||||
|
|
||||||
<div>
|
<div>
|
||||||
<Label htmlFor="destination">To</Label>
|
<Label htmlFor="destination">To</Label>
|
||||||
<SelectDropdown
|
<Input
|
||||||
|
type="text"
|
||||||
id="destination"
|
id="destination"
|
||||||
value={destination}
|
value={destination}
|
||||||
onChange={setDestination}
|
onChange={(e) => setDestination(e.target.value)}
|
||||||
options={CITIES}
|
placeholder="Airport or city"
|
||||||
placeholder="Select destination"
|
|
||||||
icon={<LocationIcon />}
|
icon={<LocationIcon />}
|
||||||
required
|
required
|
||||||
/>
|
/>
|
||||||
@@ -108,6 +115,20 @@ export default function AirlineHero() {
|
|||||||
required
|
required
|
||||||
/>
|
/>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
|
<div>
|
||||||
|
<Label htmlFor="returnDate">Return</Label>
|
||||||
|
{tripType === 'roundtrip' ? (
|
||||||
|
<DateInput
|
||||||
|
id="returnDate"
|
||||||
|
value={returnDate}
|
||||||
|
onChange={(e) => setReturnDate(e.target.value)}
|
||||||
|
required
|
||||||
|
/>
|
||||||
|
) : (
|
||||||
|
<DateInput id="returnDate" disabled />
|
||||||
|
)}
|
||||||
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
<div className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">
|
<div className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">
|
||||||
|
|||||||
@@ -1,119 +0,0 @@
|
|||||||
'use client';
|
|
||||||
|
|
||||||
import { useState, useRef, useEffect, ReactNode } from 'react';
|
|
||||||
|
|
||||||
export interface SelectOption {
|
|
||||||
value: string;
|
|
||||||
label: string;
|
|
||||||
sublabel?: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
interface SelectDropdownProps {
|
|
||||||
value: string;
|
|
||||||
onChange: (value: string) => void;
|
|
||||||
options: SelectOption[];
|
|
||||||
placeholder?: string;
|
|
||||||
icon?: ReactNode;
|
|
||||||
required?: boolean;
|
|
||||||
id?: string;
|
|
||||||
}
|
|
||||||
|
|
||||||
export default function SelectDropdown({
|
|
||||||
value,
|
|
||||||
onChange,
|
|
||||||
options,
|
|
||||||
placeholder = 'Select...',
|
|
||||||
icon,
|
|
||||||
required,
|
|
||||||
id,
|
|
||||||
}: SelectDropdownProps) {
|
|
||||||
const [open, setOpen] = useState(false);
|
|
||||||
const [filter, setFilter] = useState('');
|
|
||||||
const ref = useRef<HTMLDivElement>(null);
|
|
||||||
const inputRef = useRef<HTMLInputElement>(null);
|
|
||||||
|
|
||||||
useEffect(() => {
|
|
||||||
const handleClick = (e: MouseEvent) => {
|
|
||||||
if (ref.current && !ref.current.contains(e.target as Node)) {
|
|
||||||
setOpen(false);
|
|
||||||
setFilter('');
|
|
||||||
}
|
|
||||||
};
|
|
||||||
document.addEventListener('mousedown', handleClick);
|
|
||||||
return () => document.removeEventListener('mousedown', handleClick);
|
|
||||||
}, []);
|
|
||||||
|
|
||||||
const selectedOption = options.find((o) => o.value === value);
|
|
||||||
const filtered = options.filter(
|
|
||||||
(o) =>
|
|
||||||
o.label.toLowerCase().includes(filter.toLowerCase()) ||
|
|
||||||
o.value.toLowerCase().includes(filter.toLowerCase()) ||
|
|
||||||
o.sublabel?.toLowerCase().includes(filter.toLowerCase())
|
|
||||||
);
|
|
||||||
|
|
||||||
const handleSelect = (opt: SelectOption) => {
|
|
||||||
onChange(opt.value);
|
|
||||||
setOpen(false);
|
|
||||||
setFilter('');
|
|
||||||
};
|
|
||||||
|
|
||||||
return (
|
|
||||||
<div className="relative" ref={ref}>
|
|
||||||
<div
|
|
||||||
className="input-field flex items-center gap-2 cursor-pointer box-border"
|
|
||||||
onClick={() => {
|
|
||||||
setOpen(true);
|
|
||||||
setTimeout(() => inputRef.current?.focus(), 0);
|
|
||||||
}}
|
|
||||||
>
|
|
||||||
{icon && <span className="text-[var(--text-secondary)]">{icon}</span>}
|
|
||||||
{open ? (
|
|
||||||
<input
|
|
||||||
ref={inputRef}
|
|
||||||
type="text"
|
|
||||||
id={id}
|
|
||||||
value={filter}
|
|
||||||
onChange={(e) => setFilter(e.target.value)}
|
|
||||||
placeholder={placeholder}
|
|
||||||
className="flex-1 bg-transparent outline-none text-sm text-[var(--text-primary)]"
|
|
||||||
/>
|
|
||||||
) : (
|
|
||||||
<span className={`flex-1 text-sm ${value ? 'text-[var(--text-primary)]' : 'text-[var(--text-secondary)]'}`}>
|
|
||||||
{selectedOption ? selectedOption.label : placeholder}
|
|
||||||
</span>
|
|
||||||
)}
|
|
||||||
<svg
|
|
||||||
className={`w-4 h-4 text-[var(--text-secondary)] transition-transform ${open ? 'rotate-180' : ''}`}
|
|
||||||
fill="none"
|
|
||||||
stroke="currentColor"
|
|
||||||
viewBox="0 0 24 24"
|
|
||||||
>
|
|
||||||
<path strokeLinecap="round" strokeLinejoin="round" strokeWidth={2} d="M19 9l-7 7-7-7" />
|
|
||||||
</svg>
|
|
||||||
</div>
|
|
||||||
{open && (
|
|
||||||
<div className="absolute z-20 mt-1 w-full bg-[var(--bg-primary)] border-2 border-[var(--accent-primary)] rounded-md shadow-lg max-h-60 overflow-y-auto">
|
|
||||||
{filtered.length === 0 ? (
|
|
||||||
<div className="px-4 py-3 text-sm text-[var(--text-secondary)]">No results</div>
|
|
||||||
) : (
|
|
||||||
filtered.map((opt) => (
|
|
||||||
<div
|
|
||||||
key={opt.value}
|
|
||||||
onClick={() => handleSelect(opt)}
|
|
||||||
className={`px-4 py-2 cursor-pointer transition-colors hover:bg-[var(--accent-primary-light)] ${
|
|
||||||
opt.value === value ? 'bg-[var(--accent-primary-light)]' : ''
|
|
||||||
}`}
|
|
||||||
>
|
|
||||||
<div className="text-sm font-medium text-[var(--text-primary)]">{opt.label}</div>
|
|
||||||
{opt.sublabel && <div className="text-xs text-[var(--text-secondary)]">{opt.sublabel}</div>}
|
|
||||||
</div>
|
|
||||||
))
|
|
||||||
)}
|
|
||||||
</div>
|
|
||||||
)}
|
|
||||||
{required && !value && (
|
|
||||||
<input type="text" required className="sr-only" tabIndex={-1} value="" onChange={() => {}} />
|
|
||||||
)}
|
|
||||||
</div>
|
|
||||||
);
|
|
||||||
}
|
|
||||||
@@ -5,5 +5,3 @@ export { default as DateInput } from './DateInput';
|
|||||||
export { default as RadioGroup } from './RadioGroup';
|
export { default as RadioGroup } from './RadioGroup';
|
||||||
export { default as Dropdown, DropdownCounter } from './Dropdown';
|
export { default as Dropdown, DropdownCounter } from './Dropdown';
|
||||||
export { default as Navigation } from './Navigation';
|
export { default as Navigation } from './Navigation';
|
||||||
export { default as SelectDropdown } from './SelectDropdown';
|
|
||||||
export type { SelectOption } from './SelectDropdown';
|
|
||||||
|
|||||||
@@ -278,8 +278,6 @@
|
|||||||
padding: 12px;
|
padding: 12px;
|
||||||
transition: border-color 0.2s ease;
|
transition: border-color 0.2s ease;
|
||||||
width: 100%;
|
width: 100%;
|
||||||
min-height: 48px;
|
|
||||||
box-sizing: border-box;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
[data-mode="airline"] .input-field:focus {
|
[data-mode="airline"] .input-field:focus {
|
||||||
|
|||||||
Reference in New Issue
Block a user