Compare commits

..

14 Commits

Author SHA1 Message Date
be4313cf2a chore: cleaning shitty code 2026-01-12 10:47:29 +01:00
584a38062c small cleaning 2026-01-12 10:43:11 +01:00
4639fb7ae7 chore: updating interactions setup 2026-01-11 22:03:44 +01:00
7489ad771e chore: cleaning 2026-01-11 20:25:36 +01:00
febe22323d chore: update the test suite to be callable via makefile 2026-01-11 19:59:53 +01:00
22e6c72625 test: extra tests 2026-01-11 19:59:32 +01:00
4765c2966c features: boilerplate fixtures and stuff 2026-01-11 19:58:51 +01:00
e249f1f680 chore: dcleaning gitignore 2026-01-11 19:58:36 +01:00
73e79aaeea chore: dumping kafak just via backend 2026-01-11 19:58:27 +01:00
d294b88ef1 feature: testing setup and dummy scenarios 2026-01-11 19:58:13 +01:00
39a192e330 feature: simple npm setup 2026-01-11 19:54:09 +01:00
43cf57b34a a simp0le scaffold 2026-01-11 19:49:18 +01:00
Daniel Alves Rösel
f2271e368e 34 initial discriminator of interaction data (#38)
* feat: training pipeline + tensorboard

* tesnorboard forgot

* chore: ml basic boilerplate

* feat: naive architecture as start

* eval setup

* chore: parquet exporting of data

* chore: updating requirements necesary

* feat: separating modules and adding training logs paths

* Update experiments/ml/train.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* fix: new path for runs

* fix: undoing ai slop code

* chore: modules and reqs

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2025-12-14 18:58:42 +01:00
Daniel Alves Rösel
a1916c966c 32 refine data pipeline training data construction (#37)
* feature: modularized feature engineering for ml setup (new pipeline)

* chore: updating imports properly

* test: updating fixtures with ua and meta

* chore: migrating code ignore groups

* chore: syntax cleaning and code quality

* chore: fixing pipeline data compatability

* Update experiments/procesing/steps/session.py

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* chore: refactoring and dixing path joining

* chore: refactoring function definition to avoid reinit

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
2025-12-12 12:15:15 +01:00
20 changed files with 1220 additions and 13 deletions

3
.gitignore vendored
View File

@@ -11,3 +11,6 @@ paper/src/bib/auto
experiments/airflow/logs/*
experiments/airflow/logs/scheduler/
experiments/airflow/logs/dag_processor_manager/
tests/e2e/node_modules/**
**/auto/*.el
*.old

View File

@@ -11,46 +11,72 @@ PYTEST := $(VENV)/bin/pytest
.DEFAULT_GOAL := help
all: pdf
run.webapp:
@cd web && npm install && npm run dev
.PHONY: help
help:
@echo "pdf.build pdf.watch pdf.clean | test.backend test.e2e test.all | web.dev | install | stats.lines"
$(BUILDDIR):
mkdir -p paper/$(BUILDDIR)
pdf: $(BUILDDIR)
@echo "Concatenating source code..."
.PHONY: pdf.build
pdf.build: $(BUILDDIR)
@bash paper/concat_code.sh
@cd $(SRCDIR) && \
$(LATEXMK) -pdf -jobname=$(JOBNAME) \
-interaction=nonstopmode -file-line-error \
-outdir=../$(BUILDDIR) $(TEX)
watch: $(BUILDDIR)
.PHONY: pdf.watch
pdf.watch: $(BUILDDIR)
@cd $(SRCDIR) && \
$(LATEXMK) -pvc -pdf -jobname=$(JOBNAME) \
-interaction=nonstopmode -file-line-error \
-r ../.latexmkrc \
-outdir=../$(BUILDDIR) $(TEX)
clean:
.PHONY: pdf.clean
pdf.clean:
@cd $(SRCDIR) && \
$(LATEXMK) -C -jobname=$(JOBNAME) -outdir=../$(BUILDDIR) || true
rm -rf paper/$(BUILDDIR)/*
.PHONY: test.backend
test.backend: $(VENV)
$(PYTEST) -v
.PHONY: test.e2e
test.e2e:
@cd tests/e2e && npm install
@cd tests/e2e && npx playwright install chromium
@timeout 30 bash -c 'until curl -sf http://localhost:5000/health > /dev/null 2>&1; do sleep 1; done' || (echo "Backend not ready" && exit 1)
@timeout 30 bash -c 'until curl -sf http://localhost:3000 > /dev/null 2>&1; do sleep 1; done' || (echo "Web app not ready" && exit 1)
@cd tests/e2e && npm test
.PHONY: test.all
test.all: test.backend test.e2e
.PHONY: web.dev
web.dev:
@cd web && npm install && npm run dev
$(VENV):
python3 -m venv $(VENV)
$(PIP) install --upgrade pip
.PHONY: install
install: $(VENV)
$(PIP) install -r requirements.txt
test: $(VENV)
$(PYTEST) -v
count-lines:
.PHONY: stats.lines
stats.lines:
@find . \( -path '*/node_modules' -o -path '*/.venv' -o -path '*/venv' \) -prune -o \
\( -name "*.ts" -o -name "*.py" \) -type f -print0 | xargs -0 cat | wc -l
.PHONY: all pdf clean watch run.webapp install test
.PHONY: pdf clean watch run.webapp test count-lines all
pdf: pdf.build
clean: pdf.clean
watch: pdf.watch
run.webapp: web.dev
test: test.backend
count-lines: stats.lines
all: pdf.build

View File

@@ -1,4 +1,15 @@
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:
container_name: "PHANTOM-backend"
build:

View File

@@ -0,0 +1,115 @@
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)

View File

@@ -0,0 +1,11 @@
from .evals import evaluate
from .arch import (
XGBoostAgentClassifier,
LightGBMAgentClassifier
)
__all__ =[
'evaluate',
'XGBoostAgentClassifier',
'LightGBMAgentClassifier'
]

122
experiments/ml/arch.py Normal file
View File

@@ -0,0 +1,122 @@
# 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)]
)

103
experiments/ml/evals.py Normal file
View File

@@ -0,0 +1,103 @@
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}")

View File

@@ -0,0 +1,6 @@
torch
tensorboard
fastparquet
pyarrow
xgboost
lightgbm

137
experiments/ml/train.py Normal file
View File

@@ -0,0 +1,137 @@
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)

View File

@@ -170,3 +170,5 @@ if __name__ == '__main__':
print(f"Feature matrix: {features.shape}")
print(features.head())
print(features.info())
features.to_parquet("features.parquet")

1
tests/e2e/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""E2E test suite for PHANTOM dynamic pricing pipeline."""

17
tests/e2e/fixtures.ts Normal file
View File

@@ -0,0 +1,17 @@
import { test as base } from '@playwright/test';
type TestFixtures = {
backendUrl: string;
pricingUrl: string;
};
export const test = base.extend<TestFixtures>({
backendUrl: async ({}, use) => {
await use(process.env.BACKEND_URL || 'http://localhost:5000');
},
pricingUrl: async ({}, use) => {
await use(process.env.PRICING_PROVIDER_URL || 'http://localhost:5001');
},
});
export { expect } from '@playwright/test';

69
tests/e2e/helpers/api.ts Normal file
View File

@@ -0,0 +1,69 @@
interface PriceResponse {
price: number;
base_price: number;
markup: number;
model_version?: string;
}
export async function fetchPrice(
baseUrl: string,
productId: string,
mode: string = 'simple_surge',
sessionId?: string
): Promise<PriceResponse> {
const params = new URLSearchParams();
if (sessionId) params.set('sessionId', sessionId);
const url = `${baseUrl}/api/pricing?mode=${mode}&productId=${productId}&${params}`;
const resp = await fetch(url);
if (!resp.ok) {
throw new Error(`Price fetch failed: ${resp.status}`);
}
return resp.json();
}
export async function waitForPriceChange(
baseUrl: string,
productId: string,
baselinePrice: number,
mode: string,
sessionId?: string,
maxRetries: number = 10,
pollInterval: number = 500
): Promise<PriceResponse> {
for (let i = 0; i < maxRetries; i++) {
const priceResp = await fetchPrice(baseUrl, productId, mode, sessionId);
if (Math.abs(priceResp.price - baselinePrice) > 0.01) {
return priceResp;
}
await new Promise(r => setTimeout(r, pollInterval));
}
throw new Error(`Price did not change after ${maxRetries} retries`);
}
export async function ingestEvent(
baseUrl: string,
sessionId: string,
event: string,
productId?: string,
metadata?: Record<string, any>
): Promise<void> {
const resp = await fetch(`${baseUrl}/api/ingest`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
sessionId,
event,
productId,
timestamp: new Date().toISOString(),
metadata,
}),
});
if (!resp.ok) {
throw new Error(`Event ingest failed: ${resp.status}`);
}
}

View File

@@ -0,0 +1,219 @@
import { Page } from '@playwright/test';
export async function getSessionId(page: Page): Promise<string | null> {
const cookies = await page.context().cookies();
const sessionCookie = cookies.find(c => c.name === 'phantom_session_id');
return sessionCookie?.value || null;
}
export async function verifySessionConsistency(page: Page, expectedSessionId: string): Promise<boolean> {
const currentSessionId = await getSessionId(page);
return currentSessionId === expectedSessionId;
}
export async function createFreshSession(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
await page.context().clearCookies();
await page.goto('/');
await page.waitForLoadState('networkidle');
await page.waitForTimeout(500);
const sid = await getSessionId(page);
if (!sid) throw new Error('Session not created');
return sid;
}
interface SearchParams {
destination?: string;
checkIn?: string;
guests?: number;
rooms?: number;
origin?: string;
departure?: string;
adults?: number;
}
export async function performSearch(page: Page, params: SearchParams, storeType: 'hotel' | 'airline' = 'hotel' ): Promise<void> {
await page.waitForLoadState('networkidle');
if (storeType === 'hotel') {
const destInput = page.locator('input#destination');
await destInput.fill(params.destination || 'New York');
const checkInInput = page.locator('input#checkIn');
const checkInDate = params.checkIn || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
await checkInInput.fill(checkInDate);
const searchBtn = page.locator('button:has-text("Search Rooms")');
await searchBtn.click();
} else {
const originDropdown = page.locator('button:has-text("Select origin")').or(
page.locator('[id="origin"]').locator('button').first()
);
await originDropdown.click();
await page.waitForTimeout(200);
const originOption = page.locator(`button:has-text("${params.origin || 'JFK'}")`).first();
await originOption.click();
await page.waitForTimeout(200);
const destDropdown = page.locator('button:has-text("Select destination")').or(
page.locator('[id="destination"]').locator('button').first()
);
await destDropdown.click();
await page.waitForTimeout(200);
const destOption = page.locator(`button:has-text("${params.destination || 'LAX'}")`).first();
await destOption.click();
await page.waitForTimeout(200);
const departInput = page.locator('input#departDate');
const departDate = params.departure || new Date(Date.now() + 7 * 86400000).toISOString().split('T')[0];
await departInput.fill(departDate);
const searchBtn = page.locator('button:has-text("Search Flights")');
await searchBtn.click();
}
await page.waitForLoadState('networkidle');
}
export async function selectRandomProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string> {
await page.waitForLoadState('networkidle');
const cardClass = storeType === 'hotel' ? '.hotel-card' : '.flight-card';
const productCards = page.locator(cardClass);
const count = await productCards.count();
if (count === 0) throw new Error('No products found on listing page');
const randomIdx = Math.floor(Math.random() * count);
return randomIdx.toString();
}
export async function openProductFromListing(page: Page, productId?: string): Promise<string> {
await page.waitForLoadState('networkidle');
const hotelCards = page.locator('.hotel-card');
const flightCards = page.locator('.flight-card');
const hotelCount = await hotelCards.count();
const flightCount = await flightCards.count();
let productCards;
if (hotelCount > 0) {
productCards = hotelCards;
} else if (flightCount > 0) {
productCards = flightCards;
} else {
throw new Error('No products found on listing page');
}
const count = await productCards.count();
const randomIdx = productId ? 0 : Math.floor(Math.random() * count);
await productCards.nth(randomIdx).click();
await page.waitForLoadState('networkidle');
const url = page.url();
const match = url.match(/\/products\/([^/?]+)/);
if (!match) throw new Error('Cannot parse product ID from URL after navigation');
return match[1];
}
export async function getPriceFromDOM(page: Page): Promise<number> {
await page.waitForLoadState('networkidle');
await page.waitForSelector('.price-amount', { timeout: 15000 }).catch(() => null);
const priceSelectors = [
'.price-amount',
'.price-display',
'[data-testid="price"]',
'[data-price]',
];
for (const selector of priceSelectors) {
const priceEl = page.locator(selector).first();
if (await priceEl.count() > 0) {
const text = await priceEl.textContent();
if (!text) continue;
const match = text.match(/[\$]?\s*([\d,]+(?:\.\d{2})?)/);
if (match) {
const priceStr = match[1].replace(/,/g, '');
return parseFloat(priceStr);
}
}
}
const dataPrice = await page.locator('[data-price]').first().getAttribute('data-price').catch(() => null);
if (dataPrice) return parseFloat(dataPrice);
throw new Error('Cannot extract price from DOM');
}
export async function navigateToProduct(page: Page,productId: string,storeType: 'hotel' | 'airline' = 'hotel'): Promise<void> {
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
}
export async function viewProductViaFlow(page: Page, storeType: 'hotel' | 'airline' = 'hotel', searchParams?: SearchParams): Promise<string> {
const params = new URLSearchParams();
params.set('dateIndex', '7');
if (storeType === 'hotel') {
params.set('destination', searchParams?.destination || 'New York');
params.set('adults', '2');
params.set('rooms', '1');
} else {
params.set('origin', searchParams?.origin || 'JFK');
params.set('destination', searchParams?.destination || 'LAX');
params.set('adults', '1');
params.set('children', '0');
params.set('infants', '0');
}
await page.goto(`/products?${params.toString()}`);
await page.waitForLoadState('networkidle');
const productId = await openProductFromListing(page);
await page.waitForTimeout(500);
return productId;
}
export async function rapidViewProductViaFlow(page: Page, count: number, delayMs: number = 100, storeType: 'hotel' | 'airline' = 'hotel'): Promise<string[]> {
const productIds: string[] = [];
for (let i = 0; i < count; i++) {
const productId = await viewProductViaFlow(page, storeType);
productIds.push(productId);
await page.waitForTimeout(delayMs);
}
return productIds;
}
export async function humanLikeViewProduct(page: Page, storeType: 'hotel' | 'airline' = 'hotel'
): Promise<string> {
const productId = await viewProductViaFlow(page, storeType);
await page.hover('h1');
await page.waitForTimeout(800 + Math.random() * 400);
await page.mouse.wheel(0, 200);
await page.waitForTimeout(500 + Math.random() * 300);
const paragraphs = await page.locator('p').all();
if (paragraphs.length > 0) {
await paragraphs[0].hover();
await page.waitForTimeout(600 + Math.random() * 400);
}
return productId;
}
export async function addToCart(page: Page): Promise<void> {
const addBtn = page.locator('button:has-text("Add to Cart")');
await addBtn.click();
await page.waitForTimeout(500);
}

View File

@@ -0,0 +1,39 @@
interface InteractionEvent {
sessionId: string;
event: string;
productId?: string;
timestamp: string;
metadata?: Record<string, any>;
}
const dumpKafkaTopic = async (backendUrl: string, topic: string) => {
const resp = await fetch(`${backendUrl}/api/kafka/dump?topic=${topic}`);
if (!resp.ok) throw new Error(`Kafka dump failed: ${resp.status}`);
const { messages = [] } = await resp.json();
return messages as any[];
};
export const waitForInteractionEvent = async (
backendUrl: string,
sessionId: string,
eventType: string,
maxRetries = 10,
pollInterval = 500
): Promise<InteractionEvent | null> => {
for (let i = 0; i < maxRetries; i++) {
const msgs = await dumpKafkaTopic(backendUrl, "user-interactions");
const hit = msgs.find(m => m.sessionId === sessionId && m.event === eventType);
if (hit) return hit as InteractionEvent;
await new Promise<void>(r => setTimeout(r, pollInterval));
}
return null;
};
export const countProductViews = async (backendUrl: string, productId: string) =>
(await dumpKafkaTopic(backendUrl, "user-interactions")).reduce(
(n, m) => n + (m.productId === productId && m.event === "view_item_page" ? 1 : 0),
0
);
export const getSessionEvents = async (backendUrl: string, sessionId: string) =>
(await dumpKafkaTopic(backendUrl, "user-interactions")).filter(m => m.sessionId === sessionId);

19
tests/e2e/package.json Normal file
View File

@@ -0,0 +1,19 @@
{
"name": "e2e",
"version": "1.0.0",
"main": "index.js",
"scripts": {
"test": "playwright test",
"test:ui": "playwright test --ui",
"test:debug": "playwright test --debug"
},
"keywords": [],
"author": "",
"license": "ISC",
"description": "",
"devDependencies": {
"@playwright/test": "^1.57.0",
"@types/node": "^25.0.6",
"typescript": "^5.9.3"
}
}

View File

@@ -0,0 +1,25 @@
import { defineConfig, devices } from '@playwright/test';
export default defineConfig({
testDir: './scenarios',
fullyParallel: true,
forbidOnly: !!process.env.CI,
retries: 0,
workers: 5,
reporter: 'list',
use: {
baseURL: process.env.WEB_URL || 'http://localhost:3000',
trace: 'retain-on-failure',
screenshot: 'only-on-failure',
},
timeout: 60000,
expect: {
timeout: 10000,
},
projects: [
{
name: 'chromium',
use: { ...devices['Desktop Chrome'] },
},
],
});

View File

@@ -0,0 +1,156 @@
import { test, expect } from '../fixtures';
import {
createFreshSession,
viewProductViaFlow,
rapidViewProductViaFlow,
humanLikeViewProduct,
getPriceFromDOM,
verifySessionConsistency,
addToCart,
} from '../helpers/interactions';
import { getSessionEvents } from '../helpers/kafka';
test.describe('SessionAwarePricer E2E', () => {
const STORE_TYPE = 'hotel';
test('baseline: human-like behavior maintains base price', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
await page.waitForTimeout(1500);
const productId2 = await humanLikeViewProduct(page, STORE_TYPE);
const secondPrice = await getPriceFromDOM(page);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
expect(Math.abs(secondPrice - baselinePrice) / baselinePrice).toBeLessThan(0.1);
});
test('agent detection: rapid robot-like behavior increases price', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await page.waitForTimeout(500);
await rapidViewProductViaFlow(page, 8, 100, STORE_TYPE);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
await page.waitForTimeout(2500);
const events = await getSessionEvents(backendUrl, sessionId);
expect(events.length).toBeGreaterThanOrEqual(8);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const agentPrice = await getPriceFromDOM(page);
expect(agentPrice).toBeGreaterThan(baselinePrice);
expect((agentPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
});
test('velocity threshold: high event rate triggers detection', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
const startTime = Date.now();
await rapidViewProductViaFlow(page, 10, 80, STORE_TYPE);
const duration = (Date.now() - startTime) / 1000;
const eventsPerSec = 10 / duration;
expect(eventsPerSec).toBeGreaterThan(2.0);
await page.waitForTimeout(2000);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const agentPrice = await getPriceFromDOM(page);
expect(agentPrice).toBeGreaterThan(baselinePrice);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('cart ratio: high cart/view ratio signals intent', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await page.waitForTimeout(500);
await addToCart(page);
await page.waitForTimeout(2000);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const cartPrice = await getPriceFromDOM(page);
expect(cartPrice).toBeGreaterThanOrEqual(baselinePrice);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('mixed behavior: occasional fast actions tolerated', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId1 = await humanLikeViewProduct(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await page.waitForTimeout(1200);
await rapidViewProductViaFlow(page, 2, 150, STORE_TYPE);
await page.waitForTimeout(1500);
await humanLikeViewProduct(page, STORE_TYPE);
const finalPrice = await getPriceFromDOM(page);
expect(Math.abs(finalPrice - baselinePrice) / baselinePrice).toBeLessThan(0.3);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('session isolation: agent behavior in one session does not affect others', async ({
page,
context,
backendUrl,
}) => {
const sessionIdA = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const basePrice = await getPriceFromDOM(page);
await rapidViewProductViaFlow(page, 10, 100, STORE_TYPE);
await page.waitForTimeout(2000);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const agentPrice = await getPriceFromDOM(page);
expect(agentPrice).toBeGreaterThan(basePrice * 0.99);
const page2 = await context.newPage();
const sessionIdB = await createFreshSession(page2, STORE_TYPE);
await page2.goto(`/products/${productId}`);
await page2.waitForLoadState('networkidle');
const cleanPrice = await getPriceFromDOM(page2);
expect(Math.abs(cleanPrice - basePrice) / basePrice).toBeLessThan(0.1);
expect(sessionIdA).not.toBe(sessionIdB);
});
test('session persistence: session ID maintained across views', async ({ page }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
await viewProductViaFlow(page, STORE_TYPE);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
await viewProductViaFlow(page, STORE_TYPE);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
await viewProductViaFlow(page, STORE_TYPE);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
});

View File

@@ -0,0 +1,111 @@
import { test, expect } from '../fixtures';
import {
createFreshSession,
viewProductViaFlow,
rapidViewProductViaFlow,
getPriceFromDOM,
verifySessionConsistency,
} from '../helpers/interactions';
import { waitForInteractionEvent, countProductViews } from '../helpers/kafka';
test.describe('SimpleSurgePricer E2E', () => {
const STORE_TYPE = 'hotel';
test('baseline: initial price equals base price', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const price = await getPriceFromDOM(page);
expect(price).toBeGreaterThan(0);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('surge: rapid views trigger price increase', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
await page.waitForTimeout(2000);
const evt = await waitForInteractionEvent(backendUrl, sessionId, 'view_item_page');
expect(evt).not.toBeNull();
const viewCount = await countProductViews(backendUrl, productId);
expect(viewCount).toBeGreaterThanOrEqual(5);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const surgedPrice = await getPriceFromDOM(page);
expect(surgedPrice).toBeGreaterThan(baselinePrice);
expect((surgedPrice - baselinePrice) / baselinePrice).toBeGreaterThan(0.01);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('threshold: price unchanged below threshold', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await rapidViewProductViaFlow(page, 2, 300, STORE_TYPE);
await page.waitForTimeout(1500);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const currentPrice = await getPriceFromDOM(page);
expect(Math.abs(currentPrice - baselinePrice) / baselinePrice).toBeLessThan(0.05);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('window: surge decays after window expires', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productId = await viewProductViaFlow(page, STORE_TYPE);
const baselinePrice = await getPriceFromDOM(page);
await rapidViewProductViaFlow(page, 5, 150, STORE_TYPE);
await page.waitForTimeout(1500);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const surgedPrice = await getPriceFromDOM(page);
expect(surgedPrice).toBeGreaterThan(baselinePrice);
await page.waitForTimeout(12000);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const decayedPrice = await getPriceFromDOM(page);
expect(decayedPrice).toBeLessThan(surgedPrice);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
test('isolation: different products have independent surge', async ({ page, backendUrl }) => {
const sessionId = await createFreshSession(page, STORE_TYPE);
const productIdA = await viewProductViaFlow(page, STORE_TYPE);
const basePriceA = await getPriceFromDOM(page);
await rapidViewProductViaFlow(page, 5, 200, STORE_TYPE);
await page.waitForTimeout(2000);
await page.goto(`/products/${productIdA}`);
await page.waitForLoadState('networkidle');
const surgedPriceA = await getPriceFromDOM(page);
const productIdB = await viewProductViaFlow(page, STORE_TYPE);
const priceB = await getPriceFromDOM(page);
expect(surgedPriceA).toBeGreaterThan(basePriceA * 0.99);
expect(productIdA).not.toBe(productIdB);
expect(await verifySessionConsistency(page, sessionId)).toBeTruthy();
});
});

15
tests/e2e/tsconfig.json Normal file
View File

@@ -0,0 +1,15 @@
{
"compilerOptions": {
"target": "ES2022",
"module": "commonjs",
"lib": ["ES2022"],
"strict": true,
"esModuleInterop": true,
"skipLibCheck": true,
"forceConsistentCasingInFileNames": true,
"resolveJsonModule": true,
"types": ["node", "@playwright/test"]
},
"include": ["**/*.ts"],
"exclude": ["node_modules"]
}