Build a Bazaar Manipulation Detection API

Build a local FastAPI server that scores Bazaar items for price manipulation.

Introduction

30 Second Summary

Imagine you're trading in a busy marketplace and notice something fishy. One item's price just shot up 400% in the last hour, but almost nobody is actually buying it. Someone is probably gaming the system.

In this project, you will build a local API that loads historical Bazaar price data and instantly tells you whether any item is being manipulated. You'll query an endpoint with an item name and get back a score based on real statistical signals.

What You'll Build

You hit /check/BOOSTER_COOKIE in your browser and instantly see a manipulation score, a verdict, and the three statistical signals that produced it. You hit /check/ALL and see every item in the Bazaar ranked from most suspicious to least.

By the end of this project, you'll have:

  • A single-item endpoint that returns a manipulation score based on price Z-score, volume-price divergence, and spread anomaly signals for any Bazaar item you query.
  • A batch endpoint that scans all 1934 items at once and ranks them from most to least suspicious, with filters for minimum score and result count.
  • An event-aware detector that knows about SkyBlock events like Jerry Festival and Dark Auction, suppressing false positives when prices spike legitimately.
  • Secret Mission: Add a breakdown endpoint that shows exactly which signals triggered and how much each one contributed to the final score.

Are there any prerequisites?

You should be comfortable with basic Python concepts like variables, functions, loops, and dictionaries. No prior API experience is needed.

Before We Start

Before We Start...

Before we dive in, let's lock in what you're about to build and why it matters.

You have 1934 CSV files of Bazaar price history sitting on your machine. By the end of this project, you'll have a local API that can instantly tell you whether any item is being manipulated, backed by real statistical signals.

Set Up the Project

Before you can detect manipulation, you need a project structure with the right tools installed.

In this step, you will create a uv project, install your dependencies (FastAPI, pandas, and uvicorn), and wire up a minimal API server you can test in the browser.

In this step, get ready to:
  • Create a uv project and install FastAPI, uvicorn, and pandas.
  • Set up a data folder with your Bazaar CSV files.
  • Build and run a FastAPI health check endpoint.
Create the uv project and add dependencies
  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type Terminal (macOS) or PowerShell (Windows) and press Enter to open it.
  • Move to your Desktop by running this command:
cd ~/Desktop
  • Create your project by running this command:
uv init bazaar-detector

You should see: Initialized project `bazaar-detector` followed by the file path.

What did uv init create?

uv init created a bazaar-detector folder containing pyproject.toml (your project config), a default main.py, README.md, .python-version, and a .gitignore.

You will replace the default main.py contents with your FastAPI app shortly.

  • Enter the project and install your dependencies by running these commands:
cd bazaar-detector
uv add fastapi uvicorn pandas

uv downloads and installs the three packages. You will see resolution and installation output in the terminal.

  • Verify the packages installed correctly by running:
uv run python -c "import fastapi; import pandas; import uvicorn; print('All imports OK')"

You should see All imports OK printed in your terminal.

Seeing a ModuleNotFoundError?

  • Make sure you ran uv add fastapi uvicorn pandas from inside the bazaar-detector folder (where pyproject.toml lives).
  • Check that you used uv run python (not bare python). The uv run prefix ensures the project's virtual environment is active.

Help me debug my uv import error

  • Create the data folder by running:
mkdir data
  • Copy at least 3 of your Bazaar CSV files into the data/ folder.
  • Use Finder (macOS) or File Explorer (Windows) to drag files like DIAMOND.csv, BOOSTER_COOKIE.csv, and ENCHANTED_GOLD.csv into the data/ folder inside bazaar-detector on your Desktop.
  • Confirm your files are in place by running:
ls data/

You should see your CSV files listed (e.g. DIAMOND.csv, BOOSTER_COOKIE.csv).

Create the main application file

Now you will replace the default main.py that uv generated with your FastAPI application.

  • Press Cmd+Space (macOS) or the Windows key (Windows) to open your search bar.
  • Type PyCharm and press Enter to open it.
  • Click Open (from the welcome screen) or go to File > Open.
  • Navigate to your Desktop and select the bazaar-detector folder, then click OK.
  • In the Project panel on the left, double-click main.py to open it in the editor.
  • Select all the existing code with Cmd+A (macOS) or Ctrl+A (Windows) and delete it.
  • Paste the following code into main.py:
from pathlib import Path

import pandas as pd
from fastapi import FastAPI

# Global store for item DataFrames
item_data: dict[str, pd.DataFrame] = {}

DATA_DIR = Path("data")

app = FastAPI(
    title="Bazaar Manipulation Detector",
    description="Detects price manipulation in Hypixel SkyBlock Bazaar items",
    version="1.0.0",
)


@app.get("/")
def health_check():
    return {"status": "ok", "items_loaded": len(item_data)}

What does this code do?

  • item_data is an empty dictionary that will hold your CSV data later, keyed by item tag (like DIAMOND).
  • DATA_DIR points to the data/ folder where your CSVs live.
  • app = FastAPI(...) creates the API instance with a title and description that appear in the auto-generated docs.
  • The health_check() function responds to GET requests at / and reports how many items are loaded (zero for now).
  • Save main.py with Cmd+S (macOS) or Ctrl+S (Windows).

✔️ Awesome, I've got everything!

Great. Make sure you have saved the file before continuing.

ⓧ I'd like to double check the full code

from pathlib import Path

import pandas as pd
from fastapi import FastAPI

# Global store for item DataFrames
item_data: dict[str, pd.DataFrame] = {}

DATA_DIR = Path("data")

app = FastAPI(
    title="Bazaar Manipulation Detector",
    description="Detects price manipulation in Hypixel SkyBlock Bazaar items",
    version="1.0.0",
)


@app.get("/")
def health_check():
    return {"status": "ok", "items_loaded": len(item_data)}

Before you run the server, what do you think you will see when you visit http://127.0.0.1:8000 in your browser?

  • Open PyCharm's built-in terminal by clicking Terminal at the bottom of the PyCharm window.
  • Start the development server by running:
uv run uvicorn main:app --reload

What does this command do?

  • uv run ensures the command runs inside your project's virtual environment.
  • uvicorn main:app tells uvicorn to serve the app object from main.py.
  • --reload makes the server restart automatically whenever you save a file change.

You should see output including Uvicorn running on http://127.0.0.1:8000. The server is now running and waiting for requests.

  • Open your browser and visit http://127.0.0.1:8000.

You should see a JSON response: {"status":"ok","items_loaded":0}. The items count is zero because you have not loaded any CSVs into memory yet.

  • Now visit http://127.0.0.1:8000/docs in your browser.

You should see an interactive Swagger UI page titled Bazaar Manipulation Detector. This page auto-generates documentation for every endpoint you create. You can test your health check endpoint right here by clicking on it and hitting Try it out.

Server not starting?

  • Check you are running the command from inside the bazaar-detector folder (the one containing main.py).
  • If you see Error loading ASGI app, double-check that your main.py has no typos and that the variable is named app (not application or App).
  • If port 8000 is already in use, stop any other process on that port or run with a different port: uv run uvicorn main:app --reload --port 8001.

Help me fix my FastAPI server startup error

Your API skeleton is live and serving requests. Next up, you will load your CSV data into memory on startup so the API can actually analyze Bazaar items.

Load CSVs and Serve Raw Data

Your server is running, but it doesn't know anything about your Bazaar data yet. Right now, the item_data dictionary is empty.

Before you can detect manipulation, you need to prove your data loads correctly and is accessible through the API. In this step, you'll load your CSV files into memory when the server starts and create an endpoint that serves real price data for any item you request.

In this step, get ready to:
  • Load all CSV files into memory on server startup using a lifespan event.
  • Create an endpoint that returns the latest price data for any Bazaar item.
  • Test the endpoint using FastAPI's interactive docs.
Add a lifespan event to load CSVs on startup

FastAPI uses a lifespan pattern to run code before the server accepts any requests and after it shuts down. This is where you'll load all your CSV files into memory so every endpoint can access the data instantly.

  • In main.py, replace the entire imports section (the top 3 lines) with these updated imports:
from contextlib import asynccontextmanager
from pathlib import Path

import pandas as pd
from fastapi import FastAPI, HTTPException

What changed in the imports?

  • asynccontextmanager is the decorator you need for the lifespan function.
  • HTTPException lets you return proper HTTP error codes (like 404) when an item isn't found.
  • Now add the lifespan function directly below the DATA_DIR = Path("data") line and above the app = FastAPI(...) block by copying this code:
@asynccontextmanager
async def lifespan(app: FastAPI):
    # Load all CSVs on startup
    if DATA_DIR.exists():
        for csv_file in DATA_DIR.glob("*.csv"):
            item_tag = csv_file.stem.upper()
            df = pd.read_csv(csv_file)
            item_data[item_tag] = df
    print(f"Loaded {len(item_data)} items")
    yield
    item_data.clear()

What does this code do?

  • Everything above yield runs once at startup. It scans the data/ folder for CSV files, reads each one into a pandas DataFrame, and stores it in item_data using the filename (uppercased) as the key.
  • yield is where the server runs and handles requests.
  • Everything below yield runs on shutdown, clearing the data from memory.

Now you need to tell FastAPI to use this lifespan function.

  • Find the app = FastAPI(...) block and locate this line:
    version="1.0.0",
)
  • Add lifespan=lifespan, between the version line and the closing parenthesis, so it looks like this:
    version="1.0.0",
    lifespan=lifespan,
)
  • Save main.py.
  • Switch back to your terminal where the server is running. You should see it auto-reload and print something like Loaded 5 items (the number depends on how many CSV files you placed in data/).

Don't see Loaded N items?

  • Check that the lifespan function is defined above the app = FastAPI(...) line. Python reads top to bottom, so the function must exist before it's referenced.
  • Make sure lifespan=lifespan is inside the FastAPI() constructor call, not outside it.
  • Confirm your data/ folder contains .csv files. If the folder is empty, you'll see Loaded 0 items.

Help me figure out why my lifespan function isn't loading CSVs.

Create the check endpoint

Your data is loaded. Now you need a way to access it through the API. You'll create a path parameter endpoint where the item tag is part of the URL itself.

  • In main.py, add this endpoint below the health_check function:
@app.get("/check/{item_tag}")
def check_item(item_tag: str):
    tag = item_tag.upper()

    if tag not in item_data:
        raise HTTPException(status_code=404, detail=f"Item '{tag}' not found. Available items: {len(item_data)}")

    df = item_data[tag]
    if df.empty:
        raise HTTPException(status_code=400, detail=f"No data available for '{tag}'")

    latest = df.iloc[-1]
    return {
        "item": tag,
        "latest_price": {
            "buy_price": float(latest["buy_price"]),
            "sell_price": float(latest["sell_price"]),
            "timestamp": str(latest["timestamp"]),
        },
    }

What does this endpoint do?

  • item_tag.upper() normalizes the input so diamond, Diamond, and DIAMOND all resolve to the same item.
  • If the tag isn't found in item_data, it returns a 404 with a helpful message.
  • df.iloc[-1] grabs the last row of the DataFrame, which is the most recent data point.
  • The response packages up buy_price, sell_price, and timestamp as clean JSON.
  • Save main.py.
Test with the interactive docs

Your server has reloaded with the new endpoint. Time to see if it actually works.

  • In your browser, go to http://127.0.0.1:8000/docs to open the interactive Swagger UI.
  • You should see the new /check/{item_tag} endpoint listed alongside your health check.

Before you test, predict: if you enter DIAMOND as the item tag, what do you expect to get back?

  • Click on the GET /check/{item_tag} endpoint to expand it.
  • Click Try it out.
  • Enter DIAMOND in the item_tag field.
  • Click Execute.

You should see a JSON response with the item's latest price data, including buy_price, sell_price, and timestamp.

  • Now try entering a fake item tag like FAKE_ITEM and click Execute again.
  • You should see a 404 response with the message telling you the item wasn't found.

Getting a 404 for DIAMOND?

  • Check that your CSV file is named DIAMOND.csv (or diamond.csv). The code uppercases the filename stem, so diamond.csv becomes the key DIAMOND.
  • Check your terminal output for the Loaded N items message. If it says Loaded 0 items, your data/ folder might be empty or in the wrong location.
  • The data/ folder must be inside your bazaar-detector project root (next to main.py).

Help me debug why my /check endpoint can't find my item.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and your server is running without errors.

ⓧ I'd like to double check the full code

Your main.py should have these sections in order. Scroll up and confirm each one is present:

  • The updated imports (asynccontextmanager, Path, pandas, FastAPI, HTTPException).
  • The item_data global dict and DATA_DIR constant.
  • The lifespan function decorated with @asynccontextmanager.
  • The app = FastAPI(...) block with lifespan=lifespan included.
  • The health_check endpoint at /.
  • The check_item endpoint at /check/{item_tag}.

Your API now loads real Bazaar data and serves it on demand. Next up, you'll turn that raw data into a manipulation score that tells you whether an item's price is behaving suspiciously.

Build the Manipulation Detector

In the last step, you proved that your API loads CSV data and serves raw prices to the browser. But a raw price number doesn't answer the question you actually care about: is someone manipulating this item right now?

You need statistical signals that measure how abnormal the current price behaviour is. In this step, you'll build a detection function that computes three signals from your pre-computed data and combines them into a single manipulation score from 0 to 100.

In this step, get ready to:
  • Write a detect_manipulation() function with three statistical signals.
  • Combine the signals into a weighted score from 0 to 100.
  • Update the /check/{item_tag} endpoint to return manipulation analysis.
Write the detection function with three signals

A Z-score tells you how many standard deviations a value is from its rolling average. If an item's price suddenly jumps 4 standard deviations above normal, that's suspicious. Your CSVs already have the rolling mean and standard deviation pre-computed, so calculating a Z-score is just arithmetic.

You'll combine the price Z-score with two more signals: a volume-price divergence check and a spread anomaly check. Together, they catch different manipulation tactics.

  • In main.py, find this line:
DATA_DIR = Path("data")
  • Below that line, add the detection constants and the first signal of your function:
# --- Detection Logic ---

PRICE_Z_WEIGHT = 0.5
VOLUME_DIVERGENCE_WEIGHT = 0.3
SPREAD_ANOMALY_WEIGHT = 0.2
MANIPULATION_THRESHOLD = 60.0
EVENT_DAMPENING = 0.3


def detect_manipulation(df: pd.DataFrame) -> dict:
    latest = df.iloc[-1]

    # Signal 1: Price Z-score
    buy_mean = latest["buy_roll_mean_6"]
    buy_std = latest["buy_roll_std_6"]
    if buy_std > 0:
        price_z = abs((latest["buy_price"] - buy_mean) / buy_std)
    else:
        price_z = 0.0

What does this code do?

The five constants at the top control how signals are weighted and when to flag an item. You'll use these later when combining signals.

Signal 1 computes how far the current buy_price is from its 6-period rolling mean, measured in units of the rolling standard deviation. A value of 4.0 means the price is 4 standard deviations away from normal.

  • Directly below the Signal 1 code (still inside the function), add Signal 2 and Signal 3:
    # Signal 2: Volume-price divergence
    buy_ret = abs(latest["buy_ret"]) if pd.notna(latest["buy_ret"]) else 0.0
    current_vol = latest["buy_volume"]
    prev_vol = latest["buy_vol_lag_1"] if pd.notna(latest["buy_vol_lag_1"]) else current_vol
    if prev_vol > 0:
        vol_change = (current_vol - prev_vol) / prev_vol
    else:
        vol_change = 0.0
    if buy_ret > 0.02 and vol_change < -0.3:
        volume_divergence = min((buy_ret * 100) * abs(vol_change), 10.0)
    else:
        volume_divergence = 0.0

    # Signal 3: Spread anomaly
    spread_pct = latest["spread_pct"] if pd.notna(latest["spread_pct"]) else 0.0
    spread_vol = latest["spread_volatility"] if pd.notna(latest["spread_volatility"]) else 1.0
    if spread_vol > 0:
        spread_z = abs(spread_pct) / spread_vol
    else:
        spread_z = 0.0

What do these signals detect?

Signal 2 catches a classic manipulation pattern: the price moves sharply (buy_ret > 0.02) but volume simultaneously drops (vol_change < -0.3). Legitimate price spikes come with heavy trading volume. Manipulators move prices on thin volume.

Signal 3 checks if the spread between buy and sell prices is abnormally wide. Manipulators often inflate the buy price without matching sell orders, creating an unusually large spread.

  • Save main.py.
  • Check your terminal where the server is running. Because --reload is active, uvicorn will restart automatically. You should see the Loaded X items message reprint with no error traceback.

Seeing an IndentationError or SyntaxError?

Make sure Signal 2 and Signal 3 code is indented with exactly 4 spaces (one level inside the function body). Every line from buy_ret = through spread_z = 0.0 should be at the same indent level as latest = df.iloc[-1] above.

Also confirm that the function definition line def detect_manipulation(df: pd.DataFrame) -> dict: is at column 0 (no indentation), not nested inside another function.

Help me fix my indentation error in detect_manipulation.

Combine signals into a final score

Each raw signal uses a different scale. The price Z-score might be 0 to 10+, while the volume divergence caps at 10. You need to normalize them all to 0-100 before combining them with weights.

  • Still inside the detect_manipulation function, add this code directly below the spread_z = 0.0 line:
    # Normalize signals to 0-100 scale
    price_z_score = min(price_z / 5.0 * 100, 100.0)
    volume_div_score = min(volume_divergence / 10.0 * 100, 100.0)
    spread_anomaly_score = min(spread_z / 5.0 * 100, 100.0)

    # Weighted final score
    manipulation_score = (
        price_z_score * PRICE_Z_WEIGHT
        + volume_div_score * VOLUME_DIVERGENCE_WEIGHT
        + spread_anomaly_score * SPREAD_ANOMALY_WEIGHT
    )
    manipulation_score = round(min(manipulation_score, 100.0), 2)

    return {
        "manipulation_score": manipulation_score,
        "is_manipulated": manipulation_score >= MANIPULATION_THRESHOLD,
        "signals": {
            "price_z_score": round(price_z_score, 2),
            "volume_divergence_score": round(volume_div_score, 2),
            "spread_anomaly_score": round(spread_anomaly_score, 2),
        },
    }

What does this code do?

Each signal is divided by a maximum expected value (5.0 for Z-scores, 10.0 for volume divergence, 5.0 for spread Z) and scaled to 100. This puts all three on the same playing field.

The weighted combination gives price Z-score 50% influence, volume divergence 30%, and spread anomaly 20%. If the combined score hits 60 or above, the item is flagged as manipulated.

  • Save main.py.
  • Check your terminal. The server should reload again with no errors.

Server crashed with a NameError?

If you see NameError: name 'PRICE_Z_WEIGHT' is not defined, your constants block might be inside the function instead of at module level. The five constants (PRICE_Z_WEIGHT through EVENT_DAMPENING) must sit at column 0, above the function definition.

Help me fix the NameError in my detection function.

Update the endpoint to return manipulation analysis

Your function is complete, but nothing calls it yet. The /check/{item_tag} endpoint still returns raw price data. Time to wire it up.

  • In main.py, add a # --- App Setup --- comment on a blank line above the @asynccontextmanager decorator.
  • Scroll down to your check_item function and find these lines at the bottom of the function body:
    latest = df.iloc[-1]
    return {
        "item": tag,
        "latest_price": {
            "buy_price": float(latest["buy_price"]),
            "sell_price": float(latest["sell_price"]),
            "timestamp": str(latest["timestamp"]),
        },
    }
  • Replace that entire block (from latest = df.iloc[-1] through the closing }) with:
    result = detect_manipulation(df)
    result["item"] = tag
    result["latest_price"] = {
        "buy_price": float(df.iloc[-1]["buy_price"]),
        "sell_price": float(df.iloc[-1]["sell_price"]),
        "timestamp": str(df.iloc[-1]["timestamp"]),
    }
    return result

What changed?

Instead of building a manual response dict with raw prices, the endpoint now calls detect_manipulation(df) to get the analysis, then attaches the item tag and latest price info to the result before returning it.

  • Save main.py.

Before you test, what do you think the response will contain now? Will you still see raw prices, or something new?

  • Open your browser and visit http://127.0.0.1:8000/check/DIAMOND (or whichever item tag matches one of your CSV files).

You should see a JSON response that includes manipulation_score, is_manipulated (true or false), and a signals object with price_z_score, volume_divergence_score, and spread_anomaly_score.

Seeing a 500 Internal Server Error?

Check your terminal for the traceback. A KeyError means your CSV is missing a column that the function expects (like buy_roll_mean_6 or spread_pct). Confirm your CSV files have these pre-computed columns.

If you see NameError: name 'detect_manipulation' is not defined, the function definition must appear above the endpoint in the file. Make sure your detect_manipulation function is between DATA_DIR and the # --- App Setup --- section.

Help me debug my 500 error on the /check endpoint.

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and that your server is still running with no errors in the terminal.

ⓧ I'd like to double check the full code

Scroll up through your main.py and confirm these sections appear in order:

  • Imports (asynccontextmanager, Path, pandas, FastAPI, HTTPException) at the top.
  • Global item_data dict and DATA_DIR.
  • The # --- Detection Logic --- comment followed by five constants (PRICE_Z_WEIGHT through EVENT_DAMPENING).
  • The detect_manipulation() function with Signal 1, Signal 2, Signal 3, normalization, weighted score, and return.
  • The # --- App Setup --- comment followed by the lifespan function and app creation.
  • The health_check endpoint at /.
  • The updated check_item endpoint that calls detect_manipulation(df) and returns result.

Your API now does real analysis instead of echoing raw data. Next up, you'll handle a blind spot: items that spike during SkyBlock events like Jerry Festival aren't being manipulated, but your detector doesn't know that yet.

Fix False Positives During Events

Your manipulation detector now scores items based on price, volume, and spread signals. There's one major blind spot though.

Hypixel SkyBlock runs recurring events like Jerry Festival, Dark Auction, and Spooky Festival that cause legitimate price spikes. Without event awareness, the detector flags these real spikes as manipulation. In this step, you'll teach it the difference between manipulation and a party.

In this step, get ready to:
  • Add event detection that reads event flags from your CSV data.
  • Apply a dampening multiplier so price spikes during events don't trigger false positives.
  • Verify the API correctly reports when events are active.
See the false positive problem

Your CSV files already contain event columns that record when SkyBlock events were active at each data point. The detector currently ignores them completely.

  • Open one of your CSV files (e.g., BOOSTER_COOKIE.csv) in PyCharm or any text editor.
  • Look for columns named is_during_jerry_festival, is_during_spooky_festival, and time_delta_from_dark_auction.
  • Find any row where is_during_jerry_festival is 1 or time_delta_from_dark_auction is a small number (less than 30).

Now switch to your browser and call the endpoint for that item. Before you do, what do you think the response will say about events?

  • Visit http://127.0.0.1:8000/docs and test the /check/{item_tag} endpoint for that item.

The response has no mention of events at all. If that item's price spiked during Jerry Festival, it gets the same high manipulation score as a genuine pump-and-dump. The detector is blind to context.

Add event detection to the detector

Your CSVs already have pre-computed event flags. You need to check them and dampen the price Z-score when events are active.

  • In main.py, find this section inside detect_manipulation():
    # Normalize signals to 0-100 scale
    price_z_score = min(price_z / 5.0 * 100, 100.0)
    volume_div_score = min(volume_divergence / 10.0 * 100, 100.0)
    spread_anomaly_score = min(spread_z / 5.0 * 100, 100.0)
  • Directly below those three normalization lines, add the following event detection and dampening code:
    # Check for active events
    active_events = []
    if latest.get("is_during_season_of_jerry", 0) == 1:
        active_events.append("Season of Jerry")
    if latest.get("is_during_jerry_festival", 0) == 1:
        active_events.append("Jerry Festival")
    if latest.get("is_during_spooky_festival", 0) == 1:
        active_events.append("Spooky Festival")
    if pd.notna(latest.get("time_delta_from_dark_auction", None)):
        if abs(latest["time_delta_from_dark_auction"]) < 30:
            active_events.append("Near Dark Auction")
    if pd.notna(latest.get("time_delta_from_jacob_contest", None)):
        if abs(latest["time_delta_from_jacob_contest"]) < 15:
            active_events.append("Near Jacob Contest")

    # Apply event dampening to price Z-score
    event_suppressed = len(active_events) > 0
    if event_suppressed:
        price_z_score *= EVENT_DAMPENING

What does this code do?

This code reads five event columns from the CSV data and builds a list of any currently active events.

  • latest.get("is_during_jerry_festival", 0) safely reads the event flag, defaulting to 0 if the column is missing.
  • For time-delta columns (Dark Auction, Jacob Contest), it checks if the event happened within a threshold number of minutes (30 and 15 respectively).
  • When any event is active, EVENT_DAMPENING (0.3) multiplies the price Z-score. This reduces the biggest manipulation signal by 70% during events, since price spikes are expected.

Why dampen instead of ignore?

Events make prices volatile, but manipulation still happens during events. Dampening reduces the score without eliminating it. If someone manipulates an item during Jerry Festival, the volume-divergence and spread signals still contribute their full weight.

Now you need to include the event information in the function's response so the API can show it to callers.

  • In the same function, find the return statement at the bottom of detect_manipulation():
    return {
        "manipulation_score": manipulation_score,
        "is_manipulated": manipulation_score >= MANIPULATION_THRESHOLD,
        "signals": {
            "price_z_score": round(price_z_score, 2),
            "volume_divergence_score": round(volume_div_score, 2),
            "spread_anomaly_score": round(spread_anomaly_score, 2),
        },
    }
  • Replace the entire return statement with this updated version that includes event fields:
    return {
        "manipulation_score": manipulation_score,
        "is_manipulated": manipulation_score >= MANIPULATION_THRESHOLD,
        "signals": {
            "price_z_score": round(price_z_score, 2),
            "volume_divergence_score": round(volume_div_score, 2),
            "spread_anomaly_score": round(spread_anomaly_score, 2),
        },
        "event_suppressed": event_suppressed,
        "active_events": active_events,
    }

What changed in the return?

  • event_suppressed is a boolean. True means at least one event was active and the score was dampened.
  • active_events is a list of human-readable event names (e.g., ["Jerry Festival", "Near Dark Auction"]). Empty when no events are active.
  • Save main.py.

Your server auto-reloads thanks to the --reload flag. You should see the reload message in your terminal.

Server crashed after saving?

  • Check that the new event detection code is placed AFTER the normalization lines and BEFORE the weighted final score calculation. The order matters because price_z_score must be dampened before it's used in the weighted sum.
  • Make sure event_suppressed and active_events are defined before the return statement references them.

Help me debug my event detection code in main.py

Test the event suppression

Earlier you found a CSV row where an event was active. Now you can verify the detector handles it correctly.

  • Return to http://127.0.0.1:8000/docs in your browser.
  • Test the /check/{item_tag} endpoint with the item whose CSV had the event flag set to 1.

Before you click Execute, do you expect the event_suppressed field to be true or false for this item?

  • Click **Execute** and check the JSON response.

If the item's latest row was during an event, you should see "event_suppressed": true and the active_events list populated with the event name. The price_z_score signal is now 70% lower than it would have been without dampening.

What if event_suppressed is false?

The endpoint always reads the LATEST row of the CSV (the last line in the file). If that final row happens to be outside an event window, event_suppressed will be false even though earlier rows in the file were during events.

To confirm event detection works, check that the event_suppressed and active_events fields appear in the response at all. That proves the code runs. The actual suppression will trigger whenever the latest data point falls during an event.

Not seeing event_suppressed in the response?

  • Make sure you updated the return dictionary in detect_manipulation() to include both "event_suppressed": event_suppressed and "active_events": active_events.
  • Verify the server reloaded after you saved. You should see a reload log line in your terminal.

Why is event_suppressed missing from my API response?

✔️ Awesome, I've got everything!

Great. Double check you've saved main.py and that your server reloaded successfully.

ⓧ I'd like to double check the full code

Your main.py is now over 30 lines, so confirm each section is present in this order:

  • Imports (asynccontextmanager, Path, pandas, FastAPI, HTTPException).
  • Global item_data dict and DATA_DIR constant.
  • Weight constants (PRICE_Z_WEIGHT, VOLUME_DIVERGENCE_WEIGHT, SPREAD_ANOMALY_WEIGHT, MANIPULATION_THRESHOLD, EVENT_DAMPENING).
  • detect_manipulation() with Signal 1 (price Z-score), Signal 2 (volume-price divergence), Signal 3 (spread anomaly), normalization, event detection, dampening, weighted score, and return dict with event_suppressed and active_events.
  • lifespan() loading CSVs on startup.
  • FastAPI app instance with title, description, version.
  • health_check() at / and check_item() at /check/{item_tag}.

Your detector now distinguishes between real manipulation and event-driven price spikes. Next up, you'll add a batch endpoint that scores all 1934 items at once and ranks them by suspicion.

Add the ALL Endpoint

Your detector works well for individual items, but checking them one by one across 1934 Bazaar products is impractical.

In this step, you will upgrade the existing /check/{item_tag} endpoint so that passing ALL as the tag scans every loaded item at once and returns them ranked by suspicion score.

In this step, get ready to:
  • Add batch scanning logic that detects when the tag is ALL and iterates over every loaded item.
  • Add optional query parameters for filtering and limiting results.
  • Test the batch endpoint at /docs with various filter combinations.
Add batch scanning logic to the endpoint

Right now, the /check/{item_tag} endpoint only accepts a single item tag and returns one result. You will add a special case: when the tag is ALL, it loops through every loaded DataFrame and returns a ranked list.

FastAPI's Query class lets you add optional query parameters with descriptions that show up automatically in the interactive docs. You will use two: min_score to filter out low-suspicion items, and limit to cap how many results come back.

  • In main.py, find this import line near the top of the file:
from fastapi import FastAPI, HTTPException
  • Add Query to that import, so the line now looks like this:
from fastapi import FastAPI, HTTPException, Query
  • Now scroll down to the /check/{item_tag} endpoint. Replace the entire check_item function (from the @app.get decorator to the end of the function) with this code:
@app.get("/check/{item_tag}")
def check_item(
    item_tag: str,
    min_score: float = Query(default=0.0, description="Minimum manipulation score filter (for ALL)"),
    limit: int = Query(default=0, description="Max results to return (for ALL, 0 = no limit)"),
):
    tag = item_tag.upper()

    if tag == "ALL":
        results = []
        for name, df in item_data.items():
            if df.empty:
                continue
            result = detect_manipulation(df)
            result["item"] = name
            results.append(result)
        results.sort(key=lambda x: x["manipulation_score"], reverse=True)
        if min_score > 0:
            results = [r for r in results if r["manipulation_score"] >= min_score]
        if limit > 0:
            results = results[:limit]
        return {"total": len(results), "items": results}

What does this code do?

  • The function signature now accepts two optional Query parameters. These appear as fillable fields in the Swagger UI at /docs.
  • When the tag is ALL, it loops through every item in item_data, runs detect_manipulation() on each one, and collects the results.
  • Results are sorted by manipulation_score descending, so the most suspicious items appear first.
  • The min_score filter strips out items below a threshold, and limit caps the result count. Both default to "off" (0).
  • Immediately below the ALL block (after the return {"total": len(results), "items": results} line), keep the existing single-item code that handles individual tags. Copy it below if it was removed during the replacement:
    if tag not in item_data:
        raise HTTPException(status_code=404, detail=f"Item '{tag}' not found. Available items: {len(item_data)}")

    df = item_data[tag]
    if df.empty:
        raise HTTPException(status_code=400, detail=f"No data available for '{tag}'")

    result = detect_manipulation(df)
    result["item"] = tag
    result["latest_price"] = {
        "buy_price": float(df.iloc[-1]["buy_price"]),
        "sell_price": float(df.iloc[-1]["sell_price"]),
        "timestamp": str(df.iloc[-1]["timestamp"]),
    }
    return result

Why keep both paths in one endpoint?

Using a single /check/{item_tag} endpoint for both single items and batch scanning keeps the API simple. The learner (or any tool calling this API) just needs to remember one URL pattern.

  • Save main.py.

Before you test this, what do you expect to see when you visit /check/ALL? Think about what order the items should appear in.

  • Open your browser and visit:
http://127.0.0.1:8000/check/ALL

You should see a JSON response with a total count matching the number of items you loaded, and an items array sorted by manipulation_score descending. The most suspicious items appear at the top.

Seeing an error instead of results?

  • If you see a 422 validation error, make sure you added Query to your import line at the top of main.py.
  • If you see a 404 error, the URL is likely being matched as a single item tag. Check that the if tag == "ALL": check comes BEFORE the if tag not in item_data: check.
  • If the server crashed, check your terminal for a traceback. A common cause is a missing comma after the limit parameter in the function signature.
  • Why isn't my /check/ALL endpoint returning results?
Test with filter combinations at /docs

The min_score and limit query parameters let you narrow down the batch results. The interactive Swagger UI at /docs is the easiest way to try different combinations without editing the URL by hand.

  • Open http://127.0.0.1:8000/docs in your browser.
  • Click the GET /check/{item_tag} row to expand it.
  • Click Try it out.
  • Enter ALL in the item_tag field.
  • Enter 30 in the min_score field.
  • Enter 3 in the limit field.
  • Click Execute.

You should see at most 3 items in the response, each with a manipulation_score of 30 or higher. If none of your test items score that high, the total will be 0 and items will be an empty array.

  • Try again with min_score set to 0 and limit set to 0 to confirm you get back all loaded items with no filtering.
  • Verify that a single-item query still works by entering DIAMOND in the item_tag field and clicking Execute. You should still see the single-item response with latest_price.

What are query parameters?

Query parameters are key-value pairs that appear after a ? in a URL, like /check/ALL?min_score=30&limit=3. They let callers customize a request without changing the endpoint path.

FastAPI automatically parses them from the URL, validates their type, and documents them in the Swagger UI. The description you pass to Query() shows up as helper text in the docs.

Filters not working?

  • If min_score returns everything regardless of what you enter, check that your filter uses if min_score > 0: (not if min_score:).
  • If you don't see the min_score and limit fields in the Swagger UI, refresh the /docs page. The --reload flag should pick up changes automatically.
  • Why aren't my query parameters filtering results correctly?

✔️ Awesome, I've got everything!

Make sure you have saved main.py.

ⓧ I'd like to double check the full code

Here is the complete main.py at this point. Scroll through and confirm each section is present in order: imports, constants, detect_manipulation(), lifespan, app setup, health check, and the updated check_item endpoint.

The file has the following sections, which you built in earlier steps:

  • Imports (including Query).
  • Constants (PRICE_Z_WEIGHT, VOLUME_DIVERGENCE_WEIGHT, SPREAD_ANOMALY_WEIGHT, MANIPULATION_THRESHOLD, EVENT_DAMPENING).
  • The full detect_manipulation() function with three signals and event suppression.
  • The lifespan function that loads CSVs on startup.
  • The app instance and health_check() endpoint.
  • The updated check_item() endpoint with the ALL branch, query parameters, and single-item path.

You now have a batch scanner that ranks every Bazaar item by suspicion in a single request. Next up, you will tackle the Secret Mission and build a breakdown endpoint that explains exactly why each item got its score.

Secret mission

Signal Breakdown Endpoint

A single manipulation score is useful, but knowing which signals triggered and by how much lets you make better trading decisions. Can you build a breakdown endpoint that shows the full signal decomposition?

Clean Up Your Resources

Clean Up Your Resources

This project runs entirely locally, so there are no ongoing costs. Decide whether to keep your resources running, pause them to come back later, or delete them entirely.

Resources you used:

  • The bazaar-detector project folder on your Desktop (contains main.py, data/ folder with CSVs, and the uv-managed virtual environment).
  • The uvicorn development server (running process in your terminal).

Keep everything running

No action needed. Choose this if you're still actively building or want to keep experimenting with the API.

  • Your uvicorn server continues running in the terminal. You can keep testing endpoints at http://127.0.0.1:8000/docs.
  • Your bazaar-detector folder stays on your Desktop with all code and data intact.

Pause - I'll come back to this later

Stop the running server to free up your terminal, but keep all files so you can resume later.

  • Switch to the terminal where uvicorn is running and press Ctrl+C to stop the server.
  • Your project files remain on your Desktop. When you're ready to resume, open a terminal in the bazaar-detector folder and run uv run uvicorn main:app --reload to start the server again.

Delete - I don't want to use this again

Remove all project files and start fresh. This deletes your code, data, and the virtual environment.

  • Switch to the terminal where uvicorn is running and press Ctrl+C to stop the server.
  • Delete the project folder by running the following command from your Desktop.

macOS

cd ~/Desktop
rm -rf bazaar-detector

Windows

cd ~\Desktop
Remove-Item -Recurse -Force bazaar-detector
  • Confirm the folder is gone by running ls (macOS) or dir (Windows) on your Desktop. You should no longer see bazaar-detector listed.

That's a Wrap!

That's a Wrap!

Nice work! You just built a data-backed manipulation detection API from scratch, turning 1934 CSV files of Bazaar price history into an instant anomaly scoring system with interactive docs.

You've learned how to:

  • Build a REST API with FastAPI that loads 1934 CSV files into memory on startup using the lifespan pattern and serves auto-generated interactive docs at /docs.
  • Implement a multi-signal manipulation detector using price Z-scores, volume-price divergence, and spread anomalies combined into a weighted 0-100 score.
  • Add event-aware false positive suppression for SkyBlock events (Jerry Festival, Dark Auction, Spooky Festival) and a batch endpoint that scans all items and ranks them by suspicion score.
  • Secret Mission: Added a signal breakdown endpoint showing exactly why an item is flagged, with per-signal raw values, thresholds, trigger status, and a human-readable verdict.

Ready to quiz yourself?