Compare commits

..

14 Commits

49 changed files with 316 additions and 3106 deletions

3
.gitignore vendored
View File

@@ -11,6 +11,3 @@ 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,74 +11,46 @@ PYTEST := $(VENV)/bin/pytest
.DEFAULT_GOAL := help
.PHONY: help
help:
@echo "pdf.build pdf.watch pdf.clean | test.backend test.e2e test.all | web.dev | install | stats.lines"
all: pdf
run.webapp:
@cd web && npm install && npm run dev
$(BUILDDIR):
mkdir -p paper/$(BUILDDIR)
.PHONY: pdf.build
pdf.build: $(BUILDDIR)
pdf: $(BUILDDIR)
@echo "Concatenating source code..."
@bash paper/concat_code.sh
@cd $(SRCDIR) && \
$(LATEXMK) -pdf -jobname=$(JOBNAME) \
-interaction=nonstopmode -file-line-error \
-outdir=../$(BUILDDIR) $(TEX)
.PHONY: pdf.watch
pdf.watch: $(BUILDDIR)
watch: $(BUILDDIR)
@cd $(SRCDIR) && \
$(LATEXMK) -pvc -pdf -jobname=$(JOBNAME) \
-interaction=nonstopmode -file-line-error \
-r ../.latexmkrc \
-outdir=../$(BUILDDIR) $(TEX)
.PHONY: pdf.clean
pdf.clean:
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
@test -f tests/e2e/.env || cp tests/e2e/.env.example tests/e2e/.env
@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)
@timeout 30 bash -c 'until curl -sf http://localhost:8085/health > /dev/null 2>&1; do sleep 1; done' || (echo "Airflow 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
.PHONY: stats.lines
stats.lines:
test: $(VENV)
$(PYTEST) -v
count-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: 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
.PHONY: all pdf clean watch run.webapp install test

View File

@@ -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" />
[![Build PDF](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml/badge.svg)](https://github.com/velocitatem/PHANTOM/actions/workflows/latex.yml)
[![TPU Research Cloud](https://img.shields.io/badge/TPU%20Research%20Cloud-TRC%20supported-4285F4?logo=googlecloud&logoColor=white)](https://sites.research.google/trc/faq/)
[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-hotel.vercel.app&name=Hotel)](https://phantom-hotel.vercel.app)
[![Vercel Deploy](https://deploy-badge.vercel.app/?url=https://phantom-airline.vercel.app&name=Airline)](https://phantom-airline.vercel.app)
- https://phantom-hotel.vercel.app/
- https://phantom-airline.vercel.app/

View File

@@ -47,52 +47,53 @@ def health() -> dict:
@app.get("/api/{mode}/price/{productId}", response_model=PriceResponse)
def get_price(mode: Literal['hotel', 'airline'], productId: str, sessionId: Optional[str] = Query(None), experimentId: Optional[str] = Query(None)):
"""
THIS is the fast lookup service (mechanism).
Priority: session-keyed price > global optimal price > base price
"""
product = supabase.table(f'{mode}_products').select("metadata").eq('id', productId).execute().data[0]
if not product: raise HTTPException(404, f"Product {productId} not found")
metadata = product['metadata']
base_price = metadata.get('base_price', 100.0)
# PRIORITY 1: session-aware price (computed by Airflow worker)
if sessionId:
session_price = registry.get_session_price(sessionId, productId)
if session_price is not None:
return PriceResponse(
productId=productId,
price=session_price,
base_price=base_price,
markup=session_price/base_price,
elasticity=None,
model_version='session-aware'
)
# PRIORITY 2: global pre-computed prices (surge pricing)
# fetch pre-computed prices from registry
prices_df = registry.get_prices('latest')
if prices_df is not None:
product_price_row = prices_df[prices_df['productId'] == productId]
if not product_price_row.empty:
optimal_price = float(product_price_row['optimal_price'].iloc[0])
return PriceResponse(
productId=productId,
price=optimal_price,
base_price=base_price,
markup=optimal_price/base_price,
elasticity=None,
model_version='surge'
)
elasticity_df = registry.get_elasticity('latest')
if prices_df is None:
# fallback: no pre-computed prices available
return PriceResponse(
productId=productId,
price=base_price,
base_price=base_price,
markup=1.0,
elasticity=None
)
# lookup pre-computed price for this product
product_price_row = prices_df[prices_df['productId'] == productId]
if product_price_row.empty:
# product not in pre-computed prices, fallback to base
return PriceResponse(
productId=productId,
price=base_price,
base_price=base_price,
markup=1.0,
elasticity=None
)
optimal_price = float(product_price_row['optimal_price'].iloc[0]) # TODO: use optimal_price everywhere as aresult
# get elasticity if available
product_elasticity = None
if elasticity_df is not None:
product_elasticity_row = elasticity_df[elasticity_df['productId'] == productId]
if not product_elasticity_row.empty:
product_elasticity = float(product_elasticity_row['elasticity'].iloc[0])
# PRIORITY 3: fallback to base price
return PriceResponse(
productId=productId,
price=base_price,
price=optimal_price,
base_price=base_price,
markup=1.0,
elasticity=None,
model_version='base'
markup=optimal_price/base_price,
elasticity=product_elasticity
)
@app.get("/models")

View File

@@ -198,16 +198,12 @@ def dump_logs(
auto_offset_reset='earliest',
enable_auto_commit=False,
value_deserializer=lambda x: json.loads(x.decode('utf-8')),
consumer_timeout_ms=30000,
fetch_max_wait_ms=10000,
max_poll_records=1000
consumer_timeout_ms=5000
)
events = []
for msg in consumer:
events.append(msg.value)
if last_n and len(events) >= last_n * 2:
break
consumer.close()

View File

@@ -1,24 +1,4 @@
services:
tensorboard-rl:
image: tensorflow/tensorflow:latest
container_name: "PHANTOM-tensorboard-rl"
ports:
- "6007:6006"
volumes:
- ./sim/rl/runs:/logs
command: tensorboard --logdir=/logs --host=0.0.0.0 --port=6006
restart: unless-stopped
tensorboard-ml:
image: tensorflow/tensorflow:latest
container_name: "PHANTOM-tensorboard-ml"
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:
@@ -123,6 +103,12 @@ services:
- _AIRFLOW_WWW_USER_PASSWORD=admin
- REDIS_HOST=redis
- REDIS_PORT=6379
volumes:
- ./experiments/airflow/dags:/opt/airflow/dags
- ./experiments/airflow/logs:/opt/airflow/logs
- ./experiments/airflow/plugins:/opt/airflow/plugins
- ./experiments/procesing:/opt/airflow/procesing
- ./lib:/opt/airflow/lib
command: version
restart: "no"
@@ -143,8 +129,6 @@ services:
- AIRFLOW__CORE__LOAD_EXAMPLES=false
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
- AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
- AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
- AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
- KAFKA_HOST=kafka
- KAFKA_PORT=29092
- BACKEND_URL=http://backend:5000
@@ -154,6 +138,12 @@ services:
- REDIS_PORT=6379
ports:
- "${AIRFLOW_WEBSERVER_PORT:-8085}:8080"
volumes:
- ./experiments/airflow/dags:/opt/airflow/dags:ro
- ./experiments/airflow/logs:/opt/airflow/logs
- ./experiments/airflow/plugins:/opt/airflow/plugins:ro
- ./experiments/procesing:/opt/airflow/procesing:ro
- ./lib:/opt/airflow/lib:ro
command: webserver
restart: unless-stopped
healthcheck:
@@ -180,8 +170,6 @@ services:
- AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=true
- AIRFLOW__CORE__LOAD_EXAMPLES=false
- AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
- AIRFLOW__WEBSERVER__SECRET_KEY=${AIRFLOW_SECRET_KEY}
- AIRFLOW__API__AUTH_BACKENDS=airflow.api.auth.backend.basic_auth
- KAFKA_HOST=kafka
- KAFKA_PORT=29092
- BACKEND_URL=http://backend:5000
@@ -189,6 +177,12 @@ services:
- NEXT_PUBLIC_SUPABASE_ANON_KEY=${NEXT_PUBLIC_SUPABASE_ANON_KEY}
- REDIS_HOST=redis
- REDIS_PORT=6379
volumes:
- ./experiments/airflow/dags:/opt/airflow/dags:ro
- ./experiments/airflow/logs:/opt/airflow/logs
- ./experiments/airflow/plugins:/opt/airflow/plugins:ro
- ./experiments/procesing:/opt/airflow/procesing:ro
- ./lib:/opt/airflow/lib:ro
command: scheduler
restart: unless-stopped
healthcheck:

View File

@@ -21,10 +21,3 @@ RUN pip install --no-cache-dir \
# set airflow home
ENV AIRFLOW_HOME=/opt/airflow
COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
# create logs and plugins dirs (airflow expects them)
RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins

View File

@@ -1,41 +0,0 @@
FROM apache/airflow:2.7.3-python3.11
USER root
RUN apt-get update && apt-get install -y --no-install-recommends \
build-essential \
supervisor \
&& apt-get clean \
&& rm -rf /var/lib/apt/lists/*
USER airflow
COPY requirements.txt /tmp/requirements.txt
RUN pip install --no-cache-dir -r /tmp/requirements.txt
RUN pip install --no-cache-dir \
psycopg2-binary \
apache-airflow-providers-postgres
ENV AIRFLOW_HOME=/opt/airflow
ENV AIRFLOW__CORE__EXECUTOR=SequentialExecutor
ENV AIRFLOW__CORE__LOAD_EXAMPLES=false
ENV AIRFLOW__CORE__ENABLE_XCOM_PICKLING=true
ENV AIRFLOW__WEBSERVER__EXPOSE_CONFIG=true
# copy all code into image (standalone - no volume mounts needed)
COPY --chown=airflow:root experiments/airflow/dags ${AIRFLOW_HOME}/dags
COPY --chown=airflow:root experiments/procesing ${AIRFLOW_HOME}/procesing
COPY --chown=airflow:root lib ${AIRFLOW_HOME}/lib
RUN mkdir -p ${AIRFLOW_HOME}/logs ${AIRFLOW_HOME}/plugins
# copy entrypoint script
COPY --chown=airflow:root docker/airflow-railway-entrypoint.sh /entrypoint.sh
USER root
RUN chmod +x /entrypoint.sh
USER airflow
EXPOSE 8080
ENTRYPOINT ["/entrypoint.sh"]

View File

@@ -1,20 +0,0 @@
#!/bin/bash
set -e
# init db and create admin user on first run
airflow db migrate
# create admin user if not exists
airflow users create \
--username "${AIRFLOW_ADMIN_USER:-admin}" \
--password "${AIRFLOW_ADMIN_PASSWORD:-admin}" \
--firstname Admin \
--lastname User \
--role Admin \
--email admin@example.com || true
# start scheduler in background
airflow scheduler &
# start webserver in foreground (Railway needs one foreground process)
exec airflow webserver --port ${PORT:-8080}

View File

@@ -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)

View File

@@ -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')

View File

@@ -120,31 +120,15 @@ def apply_surge_pricing(**kwargs):
# rename demand_score to demand for pricer compatibility
data = product_features.rename(columns={'demand_score': 'demand'})
high_thresh = dag_conf.get('high_threshold', 10)
low_thresh = dag_conf.get('low_threshold', 2)
surge_mult = dag_conf.get('surge_multiplier', 1.2)
discount_mult = dag_conf.get('discount_multiplier', 0.9)
logging.info(f"Surge pricing config: high_thresh={high_thresh}, low_thresh={low_thresh}, surge_mult={surge_mult}, discount_mult={discount_mult}")
logging.info(f"Demand stats: min={data['demand'].min():.2f}, max={data['demand'].max():.2f}, mean={data['demand'].mean():.2f}")
logging.info(f"Products with high demand (>={high_thresh}): {(data['demand'] >= high_thresh).sum()}")
logging.info(f"Products with low demand (<={low_thresh}): {(data['demand'] <= low_thresh).sum()}")
surge_pricer = SimpleSurgePricer(
high_threshold=high_thresh,
low_threshold=low_thresh,
surge_multiplier=surge_mult,
discount_multiplier=discount_mult
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()
base_avg = data['base_price'].mean()
optimal_avg = data['optimal_price'].mean()
price_change_pct = ((optimal_avg - base_avg) / base_avg) * 100
logging.info(f"Price adjustment: base_avg={base_avg:.2f}, optimal_avg={optimal_avg:.2f}, change={price_change_pct:+.1f}%")
prices_df = data[['productId', 'price', 'base_price', 'optimal_price', 'demand']].rename(columns={
'price': 'current_price',
'demand': 'demand_score'

View File

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

View File

@@ -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)]
)

View File

@@ -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}")

View File

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

View File

@@ -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)

View File

@@ -2,7 +2,6 @@ from sklearn.pipeline import Pipeline
import pandas as pd
from procesing.context import PipelineContext
from procesing.providers import SupabaseProvider, BackendAPIProvider
import os
from procesing.steps import (
FetchInteractionsStep,
FetchPriceLogsStep,
@@ -13,13 +12,11 @@ from procesing.steps import (
ChunkByTimeWindowStep,
ComputeDemandForChunksStep,
AggregatePriceLogsStep,
# BuildStateSpaceStep,
FitPricingFunctionStep,
PredictPricesStep,
ComputeDemandStep,
JoinProductFeaturesStep,
ExtractSessionFeaturesStep,
JoinLabelsStep,
ValidateDataStep,
JoinProductFeaturesStep
)
from procesing.pricers import SimpleSurgePricer
@@ -109,66 +106,33 @@ def full_pipeline(context: PipelineContext,
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__':
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:
base_path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/" # os.path.join(os.path.dirname(__file__), "collected_data")
if not os.path.isdir(base_path):
return pd.DataFrame()
path = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/858c61ab-0a7f-4595-ae49-33f4365517b9/"
interactions_file = "messages(2).json"
prices_file = "messages(3).json"
files = {"user-interactions": "int.json", "price-logs": "price.json"}
file_to_read = files.get(topic, files["user-interactions"])
frames = []
data = pd.read_json(path + (interactions_file if topic == "user-interactions" else prices_file))
data = [r['payload'] for r in data['value'].to_list()]
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
context = PipelineContext(provider=ExperimentsProvider(), store_mode='hotel')
features = ml_training_pipeline(context)
print(f"Feature matrix: {features.shape}")
print(features.head())
print(features.info())
features.to_parquet("features.parquet")
product_features, prices = full_pipeline(context)
print(prices.to_string())

View File

@@ -3,46 +3,6 @@ import pandas as pd
from procesing.pricers.base import PricingFunction
def session_features_to_demand(session_features: pd.DataFrame) -> float:
"""
Map session behavioral features to demand proxy.
THIS is the critical θ̂ → D transformation for rule-based pricing.
Logic:
- High velocity → agent behavior → price up (revenue recovery)
- High cart ratio → purchase intent → price up
- Low activity → discount to convert
Returns: demand proxy score (0-20 range, higher = more demand)
"""
if session_features.empty:
return 1.0
feat = session_features.iloc[0] if len(session_features) > 0 else {}
velocity = feat.get('interaction_velocity', 0)
cart_ratio = feat.get('cart_to_view_ratio', 0)
item_views = feat.get('item_views', 0)
cart_adds = feat.get('cart_adds', 0)
# baseline demand
demand = 1.0
# agent detection: high velocity → treat as high "demand" to price up
if velocity > 2.0:
demand += 10.0 # strong agent signal
# conversion intent: cart interaction → price up
if cart_ratio > 0.1 or cart_adds > 0:
demand += 5.0
# browsing depth: many views → interest signal
if item_views > 3:
demand += min(item_views, 5.0)
return min(demand, 20.0) # cap at 20
class StaticPricer(PricingFunction):
"""Static pricing: always return fixed base prices"""
@@ -107,25 +67,21 @@ class SimpleSurgePricer(PricingFunction):
self.surge_multiplier = surge_multiplier
self.discount_multiplier = discount_multiplier
def fit(self, market_data: pd.DataFrame):
def fit(self, market_data : pd.DataFrame):
"""Extract base prices from product catalog or historical averages"""
self.base_prices = market_data['base_price'].to_numpy() if 'base_price' in market_data.columns else market_data['price'].values
return self
self.demand_history = market_data['demand'].to_numpy() if 'demand' in market_data.columns else np.zeros_like(self.base_prices)
def predict(self, state_space) -> np.ndarray:
def predict(self) -> np.ndarray:
"""
Adjust prices based on current demand using surge rules.
state_space.demand: demand proxy per product (from session features)
state_space.prices: base prices
state_space.demand: demand counts per product
state_space.prices: current prices (fallback if base_prices not set)
"""
demand = np.asarray(state_space.demand) if state_space and hasattr(state_space, 'demand') else np.array([0])
base = np.asarray(state_space.prices) if state_space and hasattr(state_space, 'prices') else self.base_prices
current_prices = self.base_prices if self.base_prices is not None else np.ones_like(demand_vector) * 99.99
demand = self.demand_history if self.demand_history is not None else np.zeros_like(current_prices)
new_prices = current_prices.copy()
if base is None:
base = np.ones(len(demand)) * 99.99
# ensure float dtype to allow multiplication by float multipliers
new_prices = base.astype(np.float64).copy()
high_mask = demand >= self.high_threshold
new_prices[high_mask] *= self.surge_multiplier

View File

@@ -18,17 +18,10 @@ class SupabaseProvider(DataProvider):
self.supabase: Client = create_client(self.supabase_url, self.supabase_key)
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("*").execute()
if not resp.data:
return 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
resp = self.supabase.table(f'{store_mode}_products').select(
"id, room_type, date_index, metadata, availability"
).execute()
return pd.DataFrame(resp.data) if resp.data else pd.DataFrame()
def fetch_experiments(self, experiment_ids: List[str]) -> pd.DataFrame:
if not experiment_ids:

View File

@@ -6,11 +6,7 @@ from procesing.steps.chunk import ChunkByTimeWindowStep
from procesing.steps.demand import ComputeDemandStep, ComputeDemandForChunksStep
from procesing.steps.elasticity import AggregatePriceLogsStep
from procesing.steps.pricing import FitPricingFunctionStep, PredictPricesStep
from procesing.steps.session import (
ExtractSessionFeaturesStep, JoinLabelsStep, ValidateDataStep,
TemporalFeatureStep, BehavioralFeatureStep, ProductFeatureStep, UserAgentFeatureStep,
_extract_features_for_session
)
from procesing.steps.session import ExtractSessionFeaturesStep, _extract_features_for_session
__all__ = [
'BaseContextStep',
@@ -29,11 +25,5 @@ __all__ = [
'FitPricingFunctionStep',
'PredictPricesStep',
'ExtractSessionFeaturesStep',
'JoinLabelsStep',
'ValidateDataStep',
'TemporalFeatureStep',
'BehavioralFeatureStep',
'ProductFeatureStep',
'UserAgentFeatureStep',
'_extract_features_for_session',
]

View File

@@ -1,7 +1,6 @@
from abc import ABC, abstractmethod
from sklearn.base import BaseEstimator, TransformerMixin
from procesing.context import PipelineContext
from typing import Any
class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
"""
@@ -17,7 +16,7 @@ class BaseContextStep(BaseEstimator, TransformerMixin, ABC):
return self
@abstractmethod
def transform(self, X) -> Any:
def transform(self, X):
"""Transform input using context. Must be implemented by subclass."""
pass

View File

@@ -7,12 +7,12 @@ class AggregatePriceLogsStep(BaseContextStep):
"""
Aggregate price logs into time windows using VECTORIZED operations.
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):
if price_logs_df.empty:
return pd.DataFrame(columns=['productId', 'price'])
return []
df = price_logs_df.copy()
ts_col = self.context.config.get('ts_col', 'ts')

View File

@@ -2,7 +2,7 @@ import pandas as pd
from procesing.steps.base import 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):
super().__init__(context)
@@ -24,10 +24,6 @@ class FetchInteractionsStep(BaseContextStep):
# drop all where page has /admin/
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
if 'metadata_dateIndex' in df.columns:
df['dateIndex'] = df['metadata_dateIndex'].astype('Int64')
@@ -42,7 +38,7 @@ class FetchInteractionsStep(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):
super().__init__(context)
@@ -54,10 +50,6 @@ class FetchPriceLogsStep(BaseContextStep):
if df.empty:
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
if self.lookback and 'ts' in df.columns:
df['ts'] = pd.to_datetime(df['ts'])

View File

@@ -1,262 +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 numpy as np
import re
from typing import Dict, Any
from typing import Optional, Dict, Any
from collections import Counter
from procesing.steps.base import BaseContextStep
EVENT_CATS = {
'page_view': ['page_view'],
'item_view': ['view_item_page', 'learn_more_about_item'],
'cart_add': ['add_item_to_cart'],
'purchase': ['purchase', 'checkout_complete'],
'hover': ['hover_over_title', 'hover_over_paragraph', 'hover_over_link', 'hover_over_button'],
# 'filter': ['filter', 'search', 'apply_filter'],
}
HEADLESS_RE = re.compile(r'HeadlessChrome|Headless|PhantomJS', re.I)
AUTOMATION_RE = re.compile(r'Selenium|Playwright|Puppeteer|WebDriver|chromedriver|geckodriver', re.I)
BROWSER_PATTERNS = [('Chrome', r'Chrome/[\d.]+'), ('Firefox', r'Firefox/[\d.]+'),
('Safari', r'Safari/[\d.]+'), ('Edge', r'Edg/[\d.]+')]
def _extract_features_for_session(session_df: pd.DataFrame, session_timeout_sec: float = 900) -> Dict[str, Any]:
"""Compute features for single session.
Args:
session_df: interaction events for this session
session_timeout_sec: max gap between events before resetting duration (default 900s = 15min)
"""
features = {}
# basic counts
features['total_interactions'] = len(session_df)
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:
if pd.isna(s): return 'Unknown'
for name, pat in BROWSER_PATTERNS:
if re.search(pat, s): return name
return 'Other'
def _apply_to_slice(df: pd.DataFrame) -> pd.DataFrame:
"""Apply feature extraction to sliding window of interactions."""
# add columns of all features at each step
new_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"]
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):
"""Vectorized time-based features: durations, velocities, gaps."""
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
# For simplicity, we return as is
return rich_dataset.copy()
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):
"""
Vectorized session feature extraction - replaces O(n^2) per-row loop.
Input: interactions_df
Output: session-level feature matrix
THIS is our main mapping from tau (trajectory) to some features vector theta - we need to do this very well. This is what will go into demand esimation.
Extract session-level behavioral features from interaction logs.
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:
if X.empty:
def transform(self, interactions_df: pd.DataFrame) -> pd.DataFrame:
if interactions_df.empty:
return pd.DataFrame()
df = X.copy()
# run all feature steps and merge on sessionId
temporal = TemporalFeatureStep(self.context).transform(df)
behavioral = BehavioralFeatureStep(self.context).transform(df)
product = ProductFeatureStep(self.context).transform(df)
ua = UserAgentFeatureStep(self.context).transform(df)
# ensure timestamp column
if 'ts' in interactions_df.columns:
interactions_df = interactions_df.copy()
interactions_df['ts'] = pd.to_datetime(interactions_df['ts'])
result = temporal
for other in [behavioral, product, ua]:
if not other.empty and 'sessionId' in other.columns:
result = result.merge(other, on='sessionId', how='left')
# group by session and compute features
session_features = []
for session_id, session_df in interactions_df.groupby('sessionId'):
new_slice = _apply_to_slice(session_df.sort_values('ts'))
session_features.append(new_slice)
# carry forward experimentId for label joining
if 'experimentId' in df.columns:
exp_map = df.groupby('sessionId')['experimentId'].first()
result = result.merge(exp_map, on='sessionId', how='left')
return result
return pd.concat(session_features, ignore_index=True)
class JoinLabelsStep(BaseContextStep):
class FilterSessionInteractionsStep(BaseContextStep):
"""
Join experiment labels to session features.
Input: (features_df, experiments_df) or features_df (fetches experiments)
Output: labeled feature matrix with is_agent column
Filter interactions DataFrame to specific session.
Input: (interactions_df, session_id)
Output: interactions_df filtered to session_id
"""
def transform(self, X : tuple) -> pd.DataFrame:
data = X;
if isinstance(data, tuple):
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
def transform(self, data: tuple) -> pd.DataFrame:
interactions_df, session_id = data
return interactions_df[interactions_df['sessionId'] == session_id].copy()

View File

@@ -144,7 +144,7 @@ def mock_price_logs_raw_kafka():
'price': 162.47,
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
'storeMode': 'hotel',
'storeMode': 'shop',
'ts': '2025-11-25T21:05:57.967Z'
}
}
@@ -157,7 +157,7 @@ def mock_price_logs_raw_kafka():
'price': 743.49,
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
'storeMode': 'hotel',
'storeMode': 'shop',
'ts': '2025-11-25T21:05:57.993Z'
}
}
@@ -170,7 +170,7 @@ def mock_price_logs_raw_kafka():
'price': 163.87,
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
'storeMode': 'hotel',
'storeMode': 'shop',
'ts': '2025-11-25T21:05:58.009Z'
}
}
@@ -183,7 +183,7 @@ def mock_price_logs_raw_kafka():
'price': 397.46,
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
'storeMode': 'hotel',
'storeMode': 'shop',
'ts': '2025-11-25T21:05:58.049Z'
}
}
@@ -196,7 +196,7 @@ def mock_price_logs_raw_kafka():
'price': 401.66,
'sessionId': 'd423ce8a-77aa-4c9a-94d4-d1adddcc3472',
'experimentId': '53aefd07-f66a-4d7f-ba8b-7ea1fc562d35',
'storeMode': 'hotel',
'storeMode': 'shop',
'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']),
'subject_name': ['Session A', 'Session B'],
'xp_human_only': [True, False],
'xp_market_mode': ['hotel', 'airline'],
'xp_market_mode': ['hotel', 'shop'],
'xp_task_id': [None, None]
})
@@ -269,13 +269,3 @@ def empty_context(empty_provider):
store_mode='hotel',
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

View File

@@ -178,49 +178,3 @@ class ModelRegistry:
return True
except:
return False
def set_session_prices(self, session_id: str, prices: Dict[str, float], ttl: int = 1800):
"""
Store prices for a specific session.
THIS is the write path for session-aware pricing.
Args:
session_id: session identifier
prices: dict of {productId: price}
ttl: time-to-live in seconds (default 30min)
"""
if not prices:
return
key = f"session:{session_id}:prices"
# use Redis hash for O(1) lookup per product
self.redis_client.hset(key, mapping={k: str(v) for k, v in prices.items()})
self.redis_client.expire(key, ttl)
def get_session_price(self, session_id: str, product_id: str) -> Optional[float]:
"""
Lookup price for (sessionId, productId).
THIS is the read path for fast provider lookup.
Returns: price or None if not found
"""
key = f"session:{session_id}:prices"
price_str = self.redis_client.hget(key, product_id)
if price_str is None:
return None
return float(price_str.decode('utf-8') if isinstance(price_str, bytes) else price_str)
def get_session_all_prices(self, session_id: str) -> Dict[str, float]:
"""Get all prices for a session."""
key = f"session:{session_id}:prices"
prices_raw = self.redis_client.hgetall(key)
if not prices_raw:
return {}
return {
(k.decode('utf-8') if isinstance(k, bytes) else k): float(v.decode('utf-8') if isinstance(v, bytes) else v)
for k, v in prices_raw.items()
}

View File

@@ -1,63 +0,0 @@
import os
from pydantic import BaseModel as Base
import json
class PayloadModel(Base):
sessionId: str
experimentId: str | None
eventName: str
page: str | None
productId: str | None
metadata: dict
storeMode: str
userAgent: str
ts: str
class ValueModel(Base):
payload: PayloadModel
encoding: str
isPayloadNull: bool
schemaId: int
size: int
class InteractionModel(Base):
partitionID: int
offset: int
timestamp: int
compression: str
isTransactional: bool
headers: list
key: dict
value: ValueModel
class Loader:
def __init__(self, src_dir: str):
self.src_dir = src_dir
self.entries = os.listdir(src_dir)
if not self.entries: raise ValueError("empty directory")
self.data = self._load_sessions()
def _is_admin_page(self, interaction: InteractionModel) -> bool:
page = interaction.value.payload.page
return page and page.startswith("/admin/")
def _load_sessions(self) -> dict:
sessions = {}
for entry in self.entries:
int_path = f"{self.src_dir}/{entry}/int.json"
raw = json.load(open(int_path))
ints = [InteractionModel(**i) for i in raw]
sessions[entry] = [i for i in ints if not self._is_admin_page(i)]
return sessions
def get_data(self) -> dict:
return self.data
def get_entries(self) -> tuple[list[str], int]:
return self.entries, len(self.entries)
if __name__ == "__main__":
DIR = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
loader = Loader(DIR)
_, n = loader.get_entries()
print(f"Loaded {n} sessions from {DIR}")

View File

@@ -1,144 +0,0 @@
from loader import Loader
from collections import defaultdict
from typing import Dict, List, Tuple, Set
import numpy as np
import graphviz
DIR = "/home/velocitatem/Documents/Projects/PHANTOM/experiments/collected_data/"
class BehaviorModel:
def __init__(self, src_dir: str = DIR):
self.loader = Loader(src_dir)
self.data = self.loader.get_data()
self.entries, self.num_entries = self.loader.get_entries()
self.mdp = None
def _state_repr(self, evt) -> str:
p = evt.value.payload
return f"{p.page or 'unk'}|{p.productId or 'none'}|{p.eventName}"
def _extract_sessions(self):
# transform raw events into sequential state trajectories per session
trajectories = []
for sid, evts in self.data.items():
if len(evts) < 2: continue
states = [self._state_repr(e) for e in sorted(evts, key=lambda x: x.timestamp)]
trajectories.append(states)
return trajectories
def _calc_transitions(self, trajectories: List[List[str]]) -> Tuple[Dict, Set]:
trans = defaultdict(lambda: defaultdict(int))
states = set()
for traj in trajectories:
for i in range(len(traj) - 1):
s, s_next = traj[i], traj[i+1]
trans[s][s_next] += 1
states.update([s, s_next])
return trans, states
def _calc_rewards(self, trajectories: List[List[str]]) -> Dict:
# reward based on session progression depth
rwd = defaultdict(list)
for traj in trajectories:
n = len(traj)
for i, s in enumerate(traj):
rwd[s].append(i / n)
return rwd
def _normalize_trans(self, counts: Dict) -> Dict:
return {s: {s_n: cnt/sum(nxt.values()) for s_n, cnt in nxt.items()}
for s, nxt in counts.items()}
def build_MDP(self) -> Dict:
trajs = self._extract_sessions()
trans_cnt, states = self._calc_transitions(trajs)
trans_prob = self._normalize_trans(trans_cnt)
state_rwd = self._calc_rewards(trajs)
state_val = {s: np.mean(r) for s, r in state_rwd.items()}
self.mdp = {
'states': sorted(list(states)),
'num_states': len(states),
'transitions': trans_prob,
'state_values': state_val,
'state_rewards': state_rwd,
'trans_counts': trans_cnt,
}
return self.mdp
def transition_prob(self, s: str, s_next: str) -> float:
if not self.mdp: raise ValueError("build MDP first")
return self.mdp['transitions'].get(s, {}).get(s_next, 0.0)
def state_value(self, s: str) -> float:
if not self.mdp: raise ValueError("build MDP first")
return self.mdp['state_values'].get(s, 0.0)
def sample_traj(self, start: str, max_len: int = 50) -> List[str]:
if not self.mdp: raise ValueError("build MDP first")
path = [start]
curr = start
for _ in range(max_len):
nxt = self.mdp['transitions'].get(curr, {})
if not nxt: break
curr = np.random.choice(list(nxt.keys()), p=list(nxt.values()))
path.append(curr)
return path
def visualize_mdp(model: BehaviorModel, threshold: float = 0.05, output: str = "mdp_graph", fmt: str = "svg", view: bool = False, export_dot: bool = False):
"""visualize MDP as directed graph using graphviz, aggregated by event type"""
if not model.mdp: raise ValueError("build MDP first")
# aggregate transitions by event type
evt_trans = defaultdict(lambda: defaultdict(float))
for s, trans in model.mdp['transitions'].items():
evt_src = s.split('|')[2]
for s_next, prob in trans.items():
evt_dst = s_next.split('|')[2]
evt_trans[evt_src][evt_dst] += prob
# normalize aggregated transitions
for evt_src in evt_trans:
total = sum(evt_trans[evt_src].values())
if total > 0:
for evt_dst in evt_trans[evt_src]:
evt_trans[evt_src][evt_dst] /= total
g = graphviz.Digraph(format=fmt)
g.attr(rankdir='LR', size='30')
g.attr('node', shape='circle', width='1', height='1')
# collect all event types
events = set(evt_trans.keys())
for trans in evt_trans.values():
events.update(trans.keys())
# add nodes for each event type
for evt in events:
g.node(evt)
# add edges above threshold
for evt_src in evt_trans:
for evt_dst, prob in evt_trans[evt_src].items():
if prob > threshold:
g.edge(evt_src, evt_dst, label=f'{prob:.2f}')
g.render(output, view=view, cleanup=True)
print(f"Saved MDP graph to {output}.{fmt}")
if export_dot:
dot_file = f"{output}.dot"
with open(dot_file, 'w') as f:
f.write(g.source)
print(f"Exported DOT source to {dot_file}")
return g
if __name__ == "__main__":
model = BehaviorModel(DIR)
mdp = model.build_MDP()
print(f"Built MDP: {mdp['num_states']} states, {sum(len(t) for t in mdp['transitions'].values())} transitions")
if not mdp['states']:
print("No states found")
exit(1)
visualize_mdp(model, threshold=0.05, output="mdp_viz", fmt="pdf", export_dot=True)

View File

@@ -1,227 +0,0 @@
from os import kill
import numpy as np
import pandas as pd
from abc import ABC, abstractmethod
from typing import Dict, Any
from environment import BusinessLogicConstraints
"""
An angine by default should have its own demand estimation mechanism from the observed observations whihc are the computer feature.
From these features we then follow the researc hstructure of q -> p with a testable and must be updatable mechanism.
"""
class BasePricingEngine(ABC):
"""base interface for all pricing engines"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
self.c = constraints
self.rng = np.random.default_rng(seed)
self.step_count = 0
@abstractmethod
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
"""compute new prices given current state and observation from environment
args:
current_prices: current price vector [N]
observation: dict containing 'price', 'demand', and possibly interaction data
returns:
new_prices: updated price vector [N]
"""
pass
@abstractmethod
def update(obs, reward, done, info):
pass
def reset(self):
"""reset engine state for new episode"""
self.step_count = 0
class WildPricingEngine(BasePricingEngine):
"""production-like pricing using online elasticity estimation via EWMA regression"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
super().__init__(constraints, seed)
# per-product unit costs (unknown to customers; known to platform)
self.unit_cost = self.rng.uniform(8.0, 40.0, size=self.c.product_catelogue_size).astype(np.float32)
# online elasticity estimate (start moderately elastic)
self.e_hat = np.full((self.c.product_catelogue_size,), -1.3, dtype=np.float32)
# EWMA state for log-log regression
self.mu_logp = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.mu_logq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.cov_pq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.var_p = np.ones(self.c.product_catelogue_size, dtype=np.float32)
# knobs typical in production
self.lr = 0.08
self.ewma = 0.05
self.eps_explore = 0.03
self.explore_scale = 0.03
def _safe_elasticity(self, e: np.ndarray) -> np.ndarray:
return np.clip(e, -5.0, -1.05)
def reset(self):
super().reset()
self.e_hat = np.full((self.c.product_catelogue_size,), -1.3, dtype=np.float32)
self.mu_logp = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.mu_logq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.cov_pq = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
self.var_p = np.ones(self.c.product_catelogue_size, dtype=np.float32)
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
self.step_count += 1
# extract demand signal (from env observation) as proxy for sales
demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
return self._update_from_demand(current_prices, demand)
def _update_from_demand(self, prices: np.ndarray, sold: np.ndarray) -> np.ndarray:
# log transforms (add 1 to handle zeros)
logp = np.log(np.clip(prices, 1e-3, None)).astype(np.float32)
logq = np.log(sold + 1.0).astype(np.float32)
# EWMA moments for per-product regression: logq ≈ a + e*logp
a = self.ewma
dp = logp - self.mu_logp
dq = logq - self.mu_logq
self.mu_logp = (1 - a) * self.mu_logp + a * logp
self.mu_logq = (1 - a) * self.mu_logq + a * logq
self.cov_pq = (1 - a) * self.cov_pq + a * (dp * dq)
self.var_p = (1 - a) * self.var_p + a * (dp * dp + 1e-6)
e_new = self.cov_pq / (self.var_p + 1e-6)
self.e_hat = self._safe_elasticity(0.9 * self.e_hat + 0.1 * e_new)
# profit-optimal price for isoelastic demand (if e < -1)
e = self.e_hat
p_star = self.unit_cost * (e / (e + 1.0))
# smooth toward p_star
new_prices = (1 - self.lr) * prices + self.lr * p_star
# exploration (small random perturbations)
if self.rng.random() < self.eps_explore:
noise = self.rng.normal(0.0, self.explore_scale, size=new_prices.shape).astype(np.float32)
new_prices = new_prices * (1.0 + noise)
# apply business guardrails (max change + bounds)
max_adj = self.c.max_price_adjustment
ratio = np.clip(new_prices / (prices + 1e-6), 1 - max_adj, 1 + max_adj)
new_prices = prices * ratio
new_prices = np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
return new_prices
class StaticPricingEngine(BasePricingEngine):
"""baseline: fixed prices throughout episode"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
super().__init__(constraints, seed)
self.fixed_prices = None
def reset(self):
super().reset()
self.fixed_prices = None
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
self.step_count += 1
if self.fixed_prices is None:
self.fixed_prices = current_prices.copy()
return self.fixed_prices.copy()
class SimpleDemandEngine(BasePricingEngine):
"""demand-driven pricing: increase price when demand rises, decrease when it falls"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
super().__init__(constraints, seed)
self.prev_demand = None
self.lr = 0.05
def reset(self):
super().reset()
self.prev_demand = None
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
self.step_count += 1
demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
if self.prev_demand is None:
self.prev_demand = demand.copy()
return current_prices.copy()
# simple rule: if demand increases, raise price; if decreases, lower price
delta_d = demand - self.prev_demand
price_adj = self.lr * np.sign(delta_d) * np.abs(delta_d) / (np.abs(self.prev_demand) + 1.0)
new_prices = current_prices * (1.0 + price_adj)
self.prev_demand = demand.copy()
# apply constraints
max_adj = self.c.max_price_adjustment
ratio = np.clip(new_prices / (current_prices + 1e-6), 1 - max_adj, 1 + max_adj)
new_prices = current_prices * ratio
return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
class RandomWalkEngine(BasePricingEngine):
"""random walk pricing with mean reversion"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
super().__init__(constraints, seed)
self.target_price = None
self.volatility = 0.02
def reset(self):
super().reset()
self.target_price = None
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
self.step_count += 1
if self.target_price is None:
self.target_price = current_prices.copy()
# random walk with mean reversion toward target
noise = self.rng.normal(0.0, self.volatility, size=current_prices.shape).astype(np.float32)
reversion = 0.01 * (self.target_price - current_prices)
new_prices = current_prices * (1.0 + noise) + reversion
# apply constraints
max_adj = self.c.max_price_adjustment
ratio = np.clip(new_prices / (current_prices + 1e-6), 1 - max_adj, 1 + max_adj)
new_prices = current_prices * ratio
return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)
class ThompsonSamplingEngine(BasePricingEngine):
"""bayesian bandit approach per product treating price as discrete action"""
def __init__(self, constraints: BusinessLogicConstraints, seed: int = 0):
super().__init__(constraints, seed)
self.n_price_levels = 5
self.alpha = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
self.beta = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
self.price_grid = None
self.last_actions = None
def reset(self):
super().reset()
self.alpha = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
self.beta = np.ones((self.c.product_catelogue_size, self.n_price_levels), dtype=np.float32)
self.price_grid = None
self.last_actions = None
def compute_prices(self, current_prices: np.ndarray, observation: Dict[str, Any]) -> np.ndarray:
self.step_count += 1
if self.price_grid is None:
# define price grid per product
lo = current_prices * 0.7
hi = current_prices * 1.3
self.price_grid = np.linspace(lo, hi, self.n_price_levels).T
demand = observation.get('demand', np.zeros(self.c.product_catelogue_size, dtype=np.float32))
# update beliefs based on last action
if self.last_actions is not None:
for i in range(self.c.product_catelogue_size):
a = self.last_actions[i]
reward = demand[i]
if reward > 0.5:
self.alpha[i, a] += reward
else:
self.beta[i, a] += 1.0
# thompson sampling: sample from posterior, pick best
new_prices = np.zeros(self.c.product_catelogue_size, dtype=np.float32)
actions = np.zeros(self.c.product_catelogue_size, dtype=int)
for i in range(self.c.product_catelogue_size):
theta = self.rng.beta(self.alpha[i], self.beta[i]).astype(np.float32)
actions[i] = int(np.argmax(theta))
new_prices[i] = self.price_grid[i, actions[i]]
self.last_actions = actions
return np.clip(new_prices, self.c.system_min_price, self.c.system_max_price).astype(np.float32)

View File

@@ -1,320 +0,0 @@
from sys import intern
import gymnasium as gym
from gymnasium import spaces
from matplotlib import interactive
import numpy as np
from dataclasses import dataclass
import pandas as pd
from typing import Callable, Optional, Dict, Any, List
# "learner" agent learning to optimize pricing
# "agent" part of environment creating demand signals that learner processes
@dataclass
class BusinessLogicConstraints():
max_price_adjustment: float = 0.30
system_max_price: float = 500.0
system_min_price: float = 1.0
product_catelogue_size: int = 100
episode_length: int = 200
sessions_per_step: int = 250
agent_share: float = 0.25
agent_recon_multiplier: float = 6.0
agent_purchase_probability: float = 0.20
coi_strength: float = 0.25
coi_threshold: float = 4.0
coi_sigmoid_temp: float = 1.25
base_human_demand: float = 0.08
base_agent_demand: float = 0.05
human_price_elasticity: float = -1.2 # assumptions here
agent_price_elasticity: float = -0.6
w_agent_loss: float = 1.0
w_volatility: float = 5.0
w_estimation_error: float = 0.25
seed: int = 7
def _sigmoid(x: np.ndarray) -> np.ndarray:
return 1.0 / (1.0 + np.exp(-x))
class CommercePlatform:
"""
This is just an extension of the state management for the environment, it does not implement anything dynamic just helps us simulate demand.
"""
def __init__(self,
product_catelogue_size: int,
max_price: float,
min_price: float,
constraints: BusinessLogicConstraints):
self.product_catelogue_size = product_catelogue_size
self.product_supply = np.random.uniform(low=10, high=50, size=(self.product_catelogue_size,))
self.max_price = max_price
self.min_price = min_price
self.constraints = constraints
self.simulation_history: List[Dict[str, Any]] = []
self._rng = np.random.default_rng(constraints.seed)
self._last_interaction_df: pd.DataFrame = pd.DataFrame()
def setup_true_demand(self, prices: np.ndarray) -> Dict[str, np.ndarray]:
# ground truth purchase propensities
p = np.clip(prices, self.min_price, self.max_price)
pn = p / self.max_price
human_prob = self.constraints.base_human_demand * (pn ** self.constraints.human_price_elasticity)
agent_prob = self.constraints.base_agent_demand * (pn ** self.constraints.agent_price_elasticity)
return {
"human_purchase_prob": np.clip(human_prob, 0.0, 0.95),
"agent_purchase_prob": np.clip(agent_prob, 0.0, 0.95)
}
def _load_behavioral_profile(actor : str, demand_forcing):
"""
This returns a markov chain with average weights which we get from interaction data of our experiments.
This defines transition probabilities between different events:
search -> view_item_price_binN: 0.7
view_item_price_binN -> add_to_cart: 0.2
we also must reweight with the demand_forcing vector or purchase probabilities per-product
"""
def _simulate_sessions(self, base_prices: np.ndarray) -> pd.DataFrame:
demand = self.setup_true_demand(base_prices)
human_pprob = demand["human_purchase_prob"]
agent_pprob = demand["agent_purchase_prob"]
events: List[Dict[str, Any]] = []
T = self.constraints.sessions_per_step
n_agent_sessions = int(round(T * self.constraints.agent_share))
n_human_sessions = T - n_agent_sessions
n_agent_ids = max(1, n_agent_sessions // 2)
session_map = {
'humans': n_human_sessions,
'agents': n_agent_ids
}
pprob_map = {
'humans': human_pprob,
'agents': agent_pprob
}
joint_events = []
for actor, n_sessions in session_map.items():
bp = _load_behavioral_profile(actor, pprob_map[actor])
counter = 0
events = []
while counter < n_sessions:
session_events = []
while len(session_events) == 0 or session_events[-1]['action'] == 'checkout':
interaction_event = bp.sample(self._rng)
interaction_event['session_id'] = f'{actor}_{counter:06d}'
# TODO any other assignments
session_events.append(interaction_event)
events.extend(session_events)
counter += 1
joint_events.extend(events)
return pd.DataFrame(joint_events)
def compute_interaction_features(self, interaction_df: pd.DataFrame) -> Dict[str, float]:
if interaction_df.empty:
return {"mean_sale_price": 0.0, "look_to_book": 0.0}
purchases = interaction_df[interaction_df["action"] == "purchase"]
mean_sale_price = float(purchases["price_paid"].mean()) if not purchases.empty else 0.0
views = float((interaction_df["action"] == "view").sum())
buys = float((interaction_df["action"] == "purchase").sum())
return {"mean_sale_price": mean_sale_price, "look_to_book": float(views / (buys + 1e-6))}
def _session_feature_table(self, df: pd.DataFrame) -> pd.DataFrame:
# TODO: adapt this
if df.empty:
return pd.DataFrame()
g = df.groupby("session_id", sort=False)
session_duration = g["t"].max() - g["t"].min()
total_interactions = g.size()
avg_time_between = g["t"].apply(lambda x: float(np.diff(np.sort(x.to_numpy())).mean()) if len(x) > 1 else 0.0)
interaction_velocity = total_interactions / (session_duration + 1e-6)
views = g.apply(lambda x: int((x["action"] == "view").sum()), include_groups=False)
cart_adds = g.apply(lambda x: int((x["action"] == "cart").sum()), include_groups=False)
purchases = g.apply(lambda x: int((x["action"] == "purchase").sum()), include_groups=False)
conversion_rate = purchases / (views + 1e-6)
is_agent = g["actor"].apply(lambda s: bool((s == "agent").any()), include_groups=False)
return pd.DataFrame({
"session_duration_sec": session_duration.astype(float),
"avg_time_between_events": avg_time_between.astype(float),
"total_interactions": total_interactions.astype(int),
"interaction_velocity": interaction_velocity.astype(float),
"item_views": views.astype(int),
"cart_adds": cart_adds.astype(int),
"purchases": purchases.astype(int),
"conversion_rate": conversion_rate.astype(float),
"is_agent": is_agent.astype(bool),
}).reset_index()
def get_interaction_data(self) -> np.ndarray:
if self._last_interaction_df.empty:
return np.array([], dtype=object)
return self._last_interaction_df.to_dict(orient="records")
class PHANTOMEnv(gym.Env):
metadata = {"render_modes": []}
def __init__(self, constraints):
super().__init__()
self.constraints = BusinessLogicConstraints()
self.action_space = spaces.Box(low=-self.constraints.max_price_adjustment,
high=self.constraints.max_price_adjustment,
shape=(self.constraints.product_catelogue_size,), dtype=np.float32)
self.observation_space = spaces.Dict({
"elasticity": spaces.Dict({
"price": spaces.Box(
low=np.full((self.constraints.product_catelogue_size,), self.constraints.system_min_price, dtype=np.float32),
high=np.full((self.constraints.product_catelogue_size,), self.constraints.system_max_price, dtype=np.float32),
dtype=np.float32),
"demand": spaces.Box(
low=np.zeros((self.constraints.product_catelogue_size,), dtype=np.float32),
high=np.full((self.constraints.product_catelogue_size,), 1e6, dtype=np.float32),
dtype=np.float32),
})
# TODO: define more features that we compute from the interaction data
})
self.commerce_platform = CommercePlatform(
product_catelogue_size=self.constraints.product_catelogue_size,
max_price=self.constraints.system_max_price,
min_price=self.constraints.system_min_price,
constraints=self.constraints)
self._rng = np.random.default_rng(self.constraints.seed)
self.t = 0
self._prev_prices: Optional[np.ndarray] = None
self.state: Dict[str, Any] = {}
def reset(self, seed: Optional[int] = None, options: Optional[dict] = None):
super().reset(seed=seed)
if seed is not None:
self._rng = np.random.default_rng(seed)
self.commerce_platform._rng = np.random.default_rng(seed)
self.t = 0
init_prices = self._rng.uniform(low=60.0, high=140.0, size=(self.constraints.product_catelogue_size,)).astype(np.float32)
self._prev_prices = init_prices.copy()
self.state = {
"elasticity": {
"price": init_prices,
"demand": np.zeros((self.constraints.product_catelogue_size,), dtype=np.float32),
}
}
return self.state, {}
def step(self, action: np.ndarray):
self.t += 1
base_prices = self.state["elasticity"]["price"].astype(np.float32)
new_prices = np.clip(base_prices * (1.0 + action.astype(np.float32)),
self.constraints.system_min_price,
self.constraints.system_max_price).astype(np.float32)
self.state["elasticity"]["price"] = new_prices
# TODO: use the commerce platform to simulate sessions
interactions_df = self.commerce_platform._simulate_sessions(new_prices)
result = self.commerce_platform.compute_interaction_features(interactions_df)
# TODO: implement COI computation to use in reward
COI = 0.0
volatility = 0.0 if self._prev_prices is None else \
float(np.mean(np.abs((new_prices - self._prev_prices) / (self._prev_prices + 1e-6))))
self._prev_prices = new_prices.copy()
revenue_observed = float(result["revenue_observed"])
agent_loss = float(result["agent_loss"])
reward = (revenue_observed
- COI
- self.constraints.w_agent_loss * agent_loss
- self.constraints.w_volatility * volatility
- self.constraints.w_estimation_error
)
terminated = self.t >= self.constraints.episode_length
info = {
"t": self.t,
"revenue_observed": revenue_observed,
"revenue_oracle": float(result["revenue_oracle"]),
"agent_loss": agent_loss,
"ux_volatility": volatility,
"mean_internal_error": err_mean,
"look_to_book": float(result["interaction_features"].get("look_to_book", 0.0)),
"mean_sale_price": float(result["interaction_features"].get("mean_sale_price", 0.0)),
"true_human_purchases_total": float(np.sum(result["true_human_demand"])),
"true_agent_purchases_total": float(np.sum(result["true_agent_purchases"])),
}
return self.state, float(reward), terminated, False, info
if __name__ == "__main__":
import matplotlib.pyplot as plt
from collections import defaultdict
runs = {}
for use_defense in (False, True):
env = PHANTOMEnv(use_defense=use_defense)
obs, _ = env.reset(seed=42)
metrics = defaultdict(list)
total_reward = 0.0
done = False
while not done:
action = env.action_space.sample()
obs, reward, done, _, info = env.step(action)
total_reward += reward
p_mean = float(np.mean(obs["elasticity"]["price"]))
q_mean = float(np.mean(obs["elasticity"]["demand"]))
p_std = float(np.std(obs["elasticity"]["price"]))
metrics['t'].append(info['t'])
metrics['price_mean'].append(p_mean)
metrics['price_std'].append(p_std)
metrics['demand_mean'].append(q_mean)
metrics['revenue_observed'].append(info['revenue_observed'])
metrics['revenue_oracle'].append(info['revenue_oracle'])
metrics['agent_loss'].append(info['agent_loss'])
metrics['ux_volatility'].append(info['ux_volatility'])
metrics['look_to_book'].append(info['look_to_book'])
metrics['reward'].append(reward)
metrics['human_purchases'].append(info['true_human_purchases_total'])
metrics['agent_purchases'].append(info['true_agent_purchases_total'])
if info['t'] % 20 == 0 or done:
print(f"defense={'ON ' if use_defense else 'OFF'} t={info['t']:03d} p={p_mean:6.2f}±{p_std:4.2f} "
f"q={q_mean:6.2f} rev={info['revenue_observed']:7.2f} oracle={info['revenue_oracle']:7.2f} "
f"loss={info['agent_loss']:6.2f} ux={info['ux_volatility']:.3f} "
f"ltb={info['look_to_book']:5.2f} r={reward:7.2f}")
runs[use_defense] = metrics
print(f"defense={'ON ' if use_defense else 'OFF'} total_reward={total_reward:.2f}\n")
fig, axes = plt.subplots(3, 3, figsize=(15, 12))
fig.suptitle('PHANTOM Environment: Defense OFF vs ON', fontsize=14, fontweight='bold')
plot_configs = [
('price_mean', 'Mean Price', 'Price'),
('demand_mean', 'Mean Demand Estimate', 'Demand'),
('revenue_observed', 'Revenue (Observed)', 'Revenue'),
('agent_loss', 'Agent Loss (Oracle - Observed)', 'Loss'),
('ux_volatility', 'UX Volatility (Price Change)', 'Volatility'),
('look_to_book', 'Look-to-Book Ratio', 'Ratio'),
('reward', 'Step Reward', 'Reward'),
('human_purchases', 'Human Purchases', 'Count'),
('agent_purchases', 'Agent Purchases', 'Count'),
]
for idx, (key, title, ylabel) in enumerate(plot_configs):
ax = axes[idx // 3, idx % 3]
for use_defense, label, color in [(False, 'No Defense', 'red'), (True, 'With Defense', 'blue')]:
m = runs[use_defense]
ax.plot(m['t'], m[key], label=label, color=color, alpha=0.7, linewidth=1.5)
ax.set_xlabel('Step')
ax.set_ylabel(ylabel)
ax.set_title(title, fontsize=10, fontweight='bold')
ax.legend(loc='best', fontsize=8)
ax.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig('phantom_env_comparison.png', dpi=150, bbox_inches='tight')
print("Plot saved to phantom_env_comparison.png")
plt.show()

View File

@@ -1,149 +0,0 @@
import numpy as np
import logging
from pathlib import Path
from typing import Dict, Type, Optional
import pickle
from torch import neg_
from torch.utils.tensorboard import SummaryWriter
from environment import PHANTOMEnv, FastTrainingConstraints, BusinessLogicConstraints
from engine import (BasePricingEngine, WildPricingEngine, StaticPricingEngine,
SimpleDemandEngine, RandomWalkEngine, ThompsonSamplingEngine)
logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s')
logger = logging.getLogger(__name__)
"""
Target training loop:
have base prices p0 from env reset and run the env step, collect reward and metrics
pass this to the pricing engine which computes the price action to take based on previous reward by learning
the new action gets passed to the step
so we alternate, step -> reward -> engine (produces price delta) -> step with price delta -> reward
to make sure the reinforcement learning inside the engine can learn we need to have trajectory of prices
CURRENT SOLUTION BELOW does not implement correct learning or updates.
"""
class EngineTrainer:
"""wrapper to run pricing engines through episodes and collect metrics"""
def __init__(self, engine: BasePricingEngine, env: PHANTOMEnv,
tb_writer: Optional[SummaryWriter] = None):
self.engine = engine
self.env = env
self.episode_metrics = []
self.tb_writer = tb_writer
self.global_step = 0
def train(self, n_episodes: int, seed: int = 42):
obs, _ = self.env.reset(seed=seed)
prices = None
for ep in range(n_episodes):
prices = self.engine.compute_prices(prices, obs)
obs, reward, done, _, info = self.env.step(prices)
self.engine.update(obs, reward, done, info)
return self
return self.episode_metrics
def evaluate(self, n_episodes: int = 10, seed: int = 100) -> Dict:
"""evaluate trained engine"""
results = {k: [] for k in ['total_reward', 'revenue_observed', 'revenue_oracle',
'agent_loss', 'ux_volatility', 'look_to_book']}
for ep in range(n_episodes):
metrics = self.run_episode(seed=seed + ep)
for k in results: results[k].append(metrics[k])
return {k: (np.mean(v), np.std(v)) for k, v in results.items()}
def make_env(fast: bool = True):
constraints = FastTrainingConstraints() if fast else BusinessLogicConstraints()
return PHANTOMEnv(constraints=constraints)
def train_engine(engine_cls: Type[BasePricingEngine], env: PHANTOMEnv,
n_episodes: int, seed: int = 42,
tb_writer: Optional[SummaryWriter] = None) -> EngineTrainer:
constraints = env.constraints
engine = engine_cls(constraints=constraints, seed=seed)
trainer = EngineTrainer(engine, env, tb_writer=tb_writer)
trainer.train(n_episodes, seed=seed)
return trainer
def save_trainer(trainer: EngineTrainer, path: Path):
"""save engine state and metrics"""
path.parent.mkdir(parents=True, exist_ok=True)
with open(path, 'wb') as f:
pickle.dump({
'engine': trainer.engine,
'metrics': trainer.episode_metrics
}, f)
logger.info(f"Saved trainer to {path}")
def load_trainer(path: Path, env: PHANTOMEnv,
tb_writer: Optional[SummaryWriter] = None) -> EngineTrainer:
"""load saved engine"""
with open(path, 'rb') as f:
data = pickle.load(f)
trainer = EngineTrainer(data['engine'], env, tb_writer=tb_writer)
trainer.episode_metrics = data['metrics']
return trainer
if __name__ == "__main__":
base_dir = Path("./runs")
base_dir.mkdir(exist_ok=True)
engines = {
"Wild": WildPricingEngine,
"Static": StaticPricingEngine,
# "SimpleDemand": SimpleDemandEngine,
"RandomWalk": RandomWalkEngine,
"ThompsonSampling": ThompsonSamplingEngine,
}
defenses = [False, True]
n_train_episodes = 50
n_eval_episodes = 10
seed = 42
fast_mode = True
logger.info(f"Training config: {n_train_episodes} episodes per engine, fast_mode={fast_mode}")
trained_trainers = {}
for engine_name, engine_cls in engines.items():
for use_defense in defenses:
defense_label = "defense_on" if use_defense else "defense_off"
run_name = f"{engine_name}_{defense_label}"
log_dir = base_dir / run_name
log_dir.mkdir(parents=True, exist_ok=True)
logger.info(f"Training {engine_name} with defense={use_defense}")
logger.info(f"Log directory: {log_dir}")
env = make_env(fast=fast_mode)
tb_writer = SummaryWriter(log_dir=str(log_dir))
trainer = train_engine(engine_cls, env, n_train_episodes, seed, tb_writer=tb_writer)
tb_writer.close()
save_path = log_dir / "trainer.pkl"
save_trainer(trainer, save_path)
trained_trainers[run_name] = (trainer, env)
logger.info("Starting evaluation")
for run_name, (trainer, env) in trained_trainers.items():
logger.info(f"Evaluating {run_name}")
results = trainer.evaluate(n_episodes=n_eval_episodes, seed=seed + 1000)
for metric, (mean, std) in results.items():
logger.info(f" {metric:20s}: {mean:10.2f} ± {std:6.2f}")
logger.info(f"Results saved to: {base_dir}")

View File

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

View File

@@ -1,17 +0,0 @@
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';

View File

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

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

@@ -1,39 +0,0 @@
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 { data = [] } = await resp.json();
return data 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);

View File

@@ -1,19 +0,0 @@
{
"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

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

View File

@@ -1,163 +0,0 @@
import { test, expect } from '../fixtures';
import {
createFreshSession,
viewProductViaFlow,
rapidViewProductViaFlow,
humanLikeViewProduct,
getPriceFromDOM,
verifySessionConsistency,
addToCart,
} from '../helpers/interactions';
import { getSessionEvents } from '../helpers/kafka';
import { runSessionPricing } from '../helpers/airflow';
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);
await runSessionPricing(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(1000);
const events = await getSessionEvents(backendUrl, sessionId);
expect(events.length).toBeGreaterThanOrEqual(8);
await runSessionPricing(STORE_TYPE);
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);
await rapidViewProductViaFlow(page, 10, 80, STORE_TYPE);
const events = await getSessionEvents(backendUrl, sessionId);
expect(events.length).toBeGreaterThanOrEqual(10);
await runSessionPricing(STORE_TYPE);
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(1000);
await humanLikeViewProduct(page, STORE_TYPE);
await runSessionPricing(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

@@ -1,118 +0,0 @@
import { test, expect } from '../fixtures';
import {
createFreshSession,
viewProductViaFlow,
rapidViewProductViaFlow,
getPriceFromDOM,
verifySessionConsistency,
} from '../helpers/interactions';
import { waitForInteractionEvent, countProductViews } from '../helpers/kafka';
import { runSurgePricing } from '../helpers/airflow';
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(1000);
const evt = await waitForInteractionEvent(backendUrl, sessionId, 'view_item_page');
expect(evt).not.toBeNull();
const viewCount = await countProductViews(backendUrl, productId);
expect(viewCount).toBeGreaterThanOrEqual(5);
await runSurgePricing(STORE_TYPE, 3, 1);
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(1000);
await runSurgePricing(STORE_TYPE, 3, 1);
await page.goto(`/products/${productId}`);
await page.waitForLoadState('networkidle');
const surgedPrice = await getPriceFromDOM(page);
expect(surgedPrice).toBeGreaterThan(baselinePrice);
await page.waitForTimeout(12000);
await runSurgePricing(STORE_TYPE, 3, 1);
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();
});
});

View File

@@ -1,15 +0,0 @@
{
"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"]
}

View File

@@ -7,7 +7,7 @@ export async function POST(req: NextRequest) {
try {
const body = await req.json();
const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
const storeMode = process.env.STORE_MODE || 'hotel';
const userAgent = req.headers.get('user-agent') || undefined;
const event: EventBase = {

View File

@@ -11,7 +11,7 @@ export async function GET(req: NextRequest) {
const productId = searchParams.get('productId');
const sessionId = searchParams.get('sessionId');
const experimentId = searchParams.get('experimentId');
const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE || 'hotel';
const storeMode = process.env.NEXT_PUBLIC_STORE_MODE || 'shop';
if (!productId) {
return NextResponse.json(
@@ -30,8 +30,6 @@ export async function GET(req: NextRequest) {
const providerUrl = process.env.PRICING_PROVIDER_URL || 'http://localhost:5001';
try {
const queryParams = new URLSearchParams();
// THIS is our entry point into the dynamic pricing where we reference the context of the sesion and experiment and ask for a price to assign to the trajectory which is expressed
// The whole pipeline gets triggered from here.
if (sessionId) queryParams.append('sessionId', sessionId);
if (experimentId) queryParams.append('experimentId', experimentId);
@@ -57,26 +55,25 @@ export async function GET(req: NextRequest) {
price = Math.round(randomBase * 100) / 100;
}
// log price to kafka asynchronously (non-blocking)
// log price to kafka for elasticity computation
if (sessionId) {
const backendUrl = process.env.BACKEND_URL || 'http://localhost:5000';
// fire and forget - don't await to avoid blocking response
fetch(`${backendUrl}/api/kafka/price-log`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
productId,
price,
sessionId,
experimentId: experimentId || undefined,
storeMode,
ts: timestamp,
}),
}).catch(err => {
if (process.env.NODE_ENV === 'development') {
console.error('[price-log-error]', err);
}
});
try {
await fetch(`${backendUrl}/api/kafka/price-log`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({
productId,
price,
sessionId,
experimentId: experimentId || undefined,
storeMode,
ts: timestamp,
}),
});
} catch (err) {
console.error('[price-log-error]', err);
}
}
if (process.env.NODE_ENV === 'development') {

View File

@@ -2,20 +2,10 @@
import { useState, FormEvent } from 'react';
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';
const CITIES: SelectOption[] = [
{ 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' },
];
type TripType = 'roundtrip' | 'oneway' | 'multicity';
const PlaneIcon = () => (
<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() {
const router = useRouter();
const [tripType, setTripType] = useState<TripType>('roundtrip');
const [origin, setOrigin] = useState('');
const [destination, setDestination] = useState('');
const [departDate, setDepartDate] = useState('');
const [returnDate, setReturnDate] = useState('');
const [passengers, setPassengers] = useState({ adults: 1, children: 0, infants: 0 });
const handleSearch = (e: FormEvent) => {
@@ -48,6 +40,8 @@ export default function AirlineHero() {
if (origin) params.set('origin', origin);
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('children', passengers.children.toString());
@@ -72,15 +66,28 @@ export default function AirlineHero() {
<div className="search-form">
<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>
<Label htmlFor="origin">From</Label>
<SelectDropdown
<Input
type="text"
id="origin"
value={origin}
onChange={setOrigin}
options={CITIES}
placeholder="Select origin"
onChange={(e) => setOrigin(e.target.value)}
placeholder="Airport or city"
icon={<PlaneIcon />}
required
/>
@@ -88,12 +95,12 @@ export default function AirlineHero() {
<div>
<Label htmlFor="destination">To</Label>
<SelectDropdown
<Input
type="text"
id="destination"
value={destination}
onChange={setDestination}
options={CITIES}
placeholder="Select destination"
onChange={(e) => setDestination(e.target.value)}
placeholder="Airport or city"
icon={<LocationIcon />}
required
/>
@@ -108,6 +115,20 @@ export default function AirlineHero() {
required
/>
</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 className="grid grid-cols-4 sm:grid-cols-3 lg:grid-cols-4 gap-4 mt-4">

View File

@@ -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>
);
}

View File

@@ -5,5 +5,3 @@ export { default as DateInput } from './DateInput';
export { default as RadioGroup } from './RadioGroup';
export { default as Dropdown, DropdownCounter } from './Dropdown';
export { default as Navigation } from './Navigation';
export { default as SelectDropdown } from './SelectDropdown';
export type { SelectOption } from './SelectDropdown';

View File

@@ -16,7 +16,7 @@ const envSchema = z.object({
// parse and validate env at module load, fail fast with descriptive errors
const parseEnv = (): Env => {
const result = envSchema.safeParse({
STORE_MODE: process.env.NEXT_PUBLIC_STORE_MODE || process.env.STORE_MODE,
STORE_MODE: process.env.STORE_MODE,
NEXT_PUBLIC_API_BASE: process.env.NEXT_PUBLIC_API_BASE,
NEXT_PUBLIC_APP_ENV: process.env.NEXT_PUBLIC_APP_ENV,
});

View File

@@ -278,8 +278,6 @@
padding: 12px;
transition: border-color 0.2s ease;
width: 100%;
min-height: 48px;
box-sizing: border-box;
}
[data-mode="airline"] .input-field:focus {