Modern price monitoring systems need to do more than tell you that a price changed.
A single abnormal listing, a scraped error, or a temporary outlier can make a traditional threshold-based detector fire an alert when nothing meaningful happened.
In this project, I built a lightweight real-time price anomaly detector in Python that combines:
- A rolling median baseline
- Median Absolute Deviation (MAD)
- Robust Z-scores
- Short-term percentage returns
- Trend confirmation
- Alert cooldowns
The goal is simple: detect meaningful price movements without overreacting to noisy observations.
Note: This project monitors retail prices from Google Shopping results through SerpApi. It is a retail-price monitoring example, not a financial exchange-data feed.
What we're building
The pipeline looks like this:
┌──────────────────────┐
│ SerpApi / Shopping │
└──────────┬───────────┘
│
▼
┌──────────────────┐
│ Price Extraction │
│ + Validation │
└────────┬─────────┘
│
▼
┌────────────────────┐
│ Rolling Price │
│ History │
└────────┬───────────┘
│
┌───────────┼───────────┐
▼ ▼ ▼
Median MAD Return %
│ │ │
└───────────┼───────────┘
▼
Robust Z-score
│
▼
Trend Confirmation
│
▼
Signal Engine
│
▼
Alert Cooldown
The implementation is intentionally small and interpretable. The complete engine is built around a single PriceAlertEngine class and a compact AnomalyResult data structure.
Why not just use standard deviation?
A common first implementation is:
price > mean + 3 * standard_deviation
The problem is that standard deviation is sensitive to extreme observations.
Suppose your historical prices are:
990, 995, 999, 1001, 1005
Then one bad observation such as:
1500
can distort the mean and standard deviation.
That can move your detection boundary away from the actual market behavior you are trying to model.
For a noisy retail environment, a more robust baseline is useful.
That's where median and Median Absolute Deviation come in.
1. Building a rolling median baseline
Instead of storing an unlimited stream of prices, the engine keeps a bounded history using Python's deque:
self.price_history = deque(maxlen=window_size)
The default window size is 12, so the detector always works from a recent local history rather than an ever-growing dataset.
The baseline is then:
baseline = statistics.median(history)
Why use a median?
Because the median is much less affected by one unusually high or low observation.
For example:
Normal prices:
999, 1000, 1001, 1002, 1003
Median:
1001
Add an extreme outlier:
999, 1000, 1001, 1002, 5000
The median remains:
1001
That makes it a useful local reference point for anomaly detection.
2. Median Absolute Deviation (MAD)
The next step is measuring how far observations normally vary around the median.
MAD is calculated as:
MAD = median(|xi - median(x)|)
The implementation is deliberately straightforward:
@staticmethod
def calculate_mad(values):
if not values:
return 0.0
median = statistics.median(values)
deviations = [
abs(value - median)
for value in values
]
return statistics.median(deviations)
The important idea is that we measure the absolute distance from the median and then take the median of those distances.
This makes MAD much less sensitive to extreme observations than standard deviation.
3. Turning MAD into a robust Z-score
Now we can calculate a robust version of the familiar Z-score:
z = 0.6745 × (x - median) / MAD
In Python:
return 0.6745 * (price - median) / mad
The implementation also handles two important edge cases:
if len(history) < 3:
return 0.0
if mad == 0:
return 0.0
That prevents the detector from making a decision before enough history exists and avoids division by zero when historical prices are identical.
Why 0.6745?
The constant scales MAD so the resulting score is approximately comparable to the familiar standard-normal Z-score.
This lets us use an intuitive threshold such as:
|robust_z| >= 3.5
without requiring the model to be trained.
4. Detecting short-term price movement
A large deviation from the baseline is useful, but it does not tell us everything.
The detector also looks at the most recent price-to-price return:
return_pct = (
(current_price - previous_price)
/ previous_price
)
For example:
Previous price = $1000
Current price = $1040
Return = 4%
That helps distinguish a statistically unusual price from a very small movement that happens to have a relatively large standardized score.
The configured default return threshold is:
return_threshold = 0.03
which corresponds to a 3% change.
5. Confirming the trend
A single spike is different from a sustained move.
The detector therefore checks the most recent observations:
recent_count = min(5, len(history))
recent_prices = history[-recent_count:]
Then it checks whether the recent prices are consistently rising:
rising = all(
recent_prices[i] <= recent_prices[i + 1]
for i in range(len(recent_prices) - 1)
)
and whether they are consistently falling:
falling = all(
recent_prices[i] >= recent_prices[i + 1]
for i in range(len(recent_prices) - 1)
)
This gives us an additional signal:
Price anomaly + rising trend
→ BULLISH_ANOMALY
Price anomaly + falling trend
→ BEARISH_ANOMALY
Price anomaly without confirmation
→ PRICE_SPIKE / PRICE_DROP
The implementation assigns higher severity to confirmed rising or falling anomalies.
6. Combining the signals
The detector starts from a safe default:
signal = "STABLE"
severity = "INFO"
Then it requires both a robust statistical signal and a short-term return threshold.
Positive anomaly:
positive_anomaly = (
robust_z >= self.z_threshold
and return_pct >= self.return_threshold
)
Negative anomaly:
negative_anomaly = (
robust_z <= -self.z_threshold
and return_pct <= -self.return_threshold
)
For positive movement:
if positive_anomaly:
if rising:
signal = "BULLISH_ANOMALY"
severity = "HIGH"
else:
signal = "PRICE_SPIKE"
severity = "MEDIUM"
For negative movement:
elif negative_anomaly:
if falling:
signal = "BEARISH_ANOMALY"
severity = "HIGH"
else:
signal = "PRICE_DROP"
severity = "MEDIUM"
This is an important design choice.
We are not saying:
"The Z-score is high, therefore alert."
We are saying:
"The price is statistically unusual, the recent movement is large enough, and the short-term context tells us what kind of anomaly it is."
The complete evaluation logic follows this sequence and only appends the new observation after evaluating it against the previous history.
Getting real retail prices with SerpApi
The project uses SerpApi to query Google Shopping results.
The request parameters are:
params = {
"engine": "google_shopping",
"q": f"{asset_name} price",
"hl": "en",
"gl": "us",
}
Instead of blindly trusting the first listing, the engine checks up to the first ten shopping results:
for item in shopping_results[:10]:
raw_price = item.get("price")
It extracts numeric values, ignores unusable observations, and then calculates the median of the usable prices:
market_price = statistics.median(prices)
That final median becomes the market observation used by the anomaly detector.
This extra median step is useful because a single seller listing should not automatically become the entire market price signal.
The result object
Each evaluation returns a structured result:
@dataclass
class AnomalyResult:
price: float
baseline: float
return_pct: float
robust_z: float
signal: str
severity: str
That gives downstream code everything it needs to display, log, store, or route the event elsewhere.
For example:
Price : $1049.00
Baseline : $999.00
Return : +5.01%
Robust Z : +4.12
Signal : PRICE_SPIKE
Severity : MEDIUM
Avoiding alert spam
A detector that sends the same alert every polling cycle becomes annoying very quickly.
The engine therefore implements a simple cooldown:
def should_emit_alert(self):
now = time.time()
if self.last_alert_time is None:
self.last_alert_time = now
return True
if now - self.last_alert_time >= self.cooldown_seconds:
self.last_alert_time = now
return True
return False
With a cooldown of five minutes:
cooldown_seconds = 300
the system can recognize repeated anomalies without emitting an alert every few seconds.
This separates two ideas that are easy to confuse:
Detection ≠ Notification
The detector may identify an anomaly while the notification layer decides whether it is time to emit another alert.
Running the monitor
The monitor is implemented as a polling loop:
while True:
market_price = self.fetch_live_market_price(asset_name)
if market_price is not None:
evaluation = self.evaluate_market_anomaly(market_price)
if evaluation.signal != "STABLE" and evaluation.signal != "INITIALIZING":
if self.should_emit_alert():
self.display_result(evaluation)
time.sleep(self.poll_interval)
The loop also supports a controlled max_cycles value, which is useful for local verification and testing.
Complete minimal setup
save this code snippet as price_alert_engine.py
import os
import sys
import time
import statistics
from dataclasses import dataclass
from datetime import datetime, timezone
from collections import deque
from typing import Optional
import serpapi
@dataclass
class AnomalyResult:
price: float
baseline: float
return_pct: float
robust_z: float
signal: str
severity: str
class PriceAlertEngine:
"""
Real-time price monitoring and anomaly detection engine.
Detection strategy:
1. Rolling median baseline
2. Median Absolute Deviation (MAD)
3. Robust Z-score
4. Short-term percentage return
5. Trend confirmation
6. Alert cooldown
This is intentionally lightweight and interpretable.
"""
def __init__(
self,
api_key: str,
window_size: int = 12,
poll_interval: int = 60,
z_threshold: float = 3.5,
return_threshold: float = 0.03,
cooldown_seconds: int = 300,
):
if not api_key:
raise ValueError("Missing SERPAPI_API_KEY.")
self.client = serpapi.Client(api_key=api_key)
self.window_size = window_size
self.poll_interval = poll_interval
self.z_threshold = z_threshold
self.return_threshold = return_threshold
self.cooldown_seconds = cooldown_seconds
self.price_history = deque(maxlen=window_size)
self.last_alert_time: Optional[float] = None
# MARKET DATA
def fetch_live_market_price(self, asset_name: str) -> Optional[float]:
"""
Fetches the first usable Google Shopping price observation.
Note:
This is a retail-price monitoring example, not an exchange feed.
"""
print(
f"[{datetime.now().strftime('%H:%M:%S')}] "
f"[DATA] Fetching price for {asset_name}..."
)
params = {
"engine": "google_shopping",
"q": f"{asset_name} price",
"hl": "en",
"gl": "us",
}
try:
results = self.client.search(params)
shopping_results = results.get("shopping_results", [])
if not shopping_results:
print("[DATA] No shopping results returned.")
return None
# Search several results instead of trusting item #1 blindly.
prices = []
for item in shopping_results[:10]:
raw_price = item.get("price")
if not raw_price:
continue
cleaned = "".join(
char
for char in str(raw_price)
if char.isdigit() or char == "."
)
try:
value = float(cleaned)
if value > 0:
prices.append(value)
except ValueError:
continue
if not prices:
print("[DATA] No valid prices could be parsed.")
return None
# Median reduces the impact of one abnormal seller listing.
market_price = statistics.median(prices)
return market_price
except Exception as exc:
print(f"[ERROR] Market data request failed: {exc}")
return None
# STATISTICS
@staticmethod
def calculate_mad(values):
"""
Median Absolute Deviation.
MAD is much more resistant to extreme observations
than standard deviation.
"""
if not values:
return 0.0
median = statistics.median(values)
deviations = [
abs(value - median)
for value in values
]
return statistics.median(deviations)
def calculate_robust_z_score(self, price: float, history):
"""
Robust Z-score:
z = 0.6745 * (x - median) / MAD
0.6745 makes the score approximately comparable
to the familiar standard-normal z-score.
"""
if len(history) < 3:
return 0.0
median = statistics.median(history)
mad = self.calculate_mad(history)
# Prevent division by zero when all historical values are equal.
if mad == 0:
return 0.0
return 0.6745 * (price - median) / mad
# ANOMALY DETECTION
def evaluate_market_anomaly(
self,
current_price: float,
) -> AnomalyResult:
# Not enough observations yet.
if len(self.price_history) < 3:
self.price_history.append(current_price)
baseline = statistics.median(self.price_history)
return AnomalyResult(
price=current_price,
baseline=baseline,
return_pct=0.0,
robust_z=0.0,
signal="INITIALIZING",
severity="INFO",
)
history = list(self.price_history)
baseline = statistics.median(history)
robust_z = self.calculate_robust_z_score(
current_price,
history,
)
previous_price = history[-1]
return_pct = (
(current_price - previous_price)
/ previous_price
)
# Trend confirmation
recent_count = min(5, len(history))
recent_prices = history[-recent_count:]
rising = all(
recent_prices[i] <= recent_prices[i + 1]
for i in range(len(recent_prices) - 1)
)
falling = all(
recent_prices[i] >= recent_prices[i + 1]
for i in range(len(recent_prices) - 1)
)
# Signal engine
signal = "STABLE"
severity = "INFO"
positive_anomaly = (
robust_z >= self.z_threshold
and return_pct >= self.return_threshold
)
negative_anomaly = (
robust_z <= -self.z_threshold
and return_pct <= -self.return_threshold
)
if positive_anomaly:
if rising:
signal = "BULLISH_ANOMALY"
severity = "HIGH"
else:
signal = "PRICE_SPIKE"
severity = "MEDIUM"
elif negative_anomaly:
if falling:
signal = "BEARISH_ANOMALY"
severity = "HIGH"
else:
signal = "PRICE_DROP"
severity = "MEDIUM"
# Add observation after calculating against previous history.
self.price_history.append(current_price)
return AnomalyResult(
price=current_price,
baseline=baseline,
return_pct=return_pct,
robust_z=robust_z,
signal=signal,
severity=severity,
)
# ALERT CONTROL
def should_emit_alert(self) -> bool:
now = time.time()
if self.last_alert_time is None:
self.last_alert_time = now
return True
if now - self.last_alert_time >= self.cooldown_seconds:
self.last_alert_time = now
return True
return False
# DISPLAY
@staticmethod
def display_result(result: AnomalyResult):
timestamp = datetime.now(timezone.utc).strftime(
"%Y-%m-%d %H:%M:%S UTC"
)
print("\n" + "-" * 70)
print(f"Timestamp : {timestamp}")
print(f"Price : ${result.price:.2f}")
print(f"Baseline : ${result.baseline:.2f}")
print(f"Return : {result.return_pct * 100:+.2f}%")
print(f"Robust Z : {result.robust_z:+.2f}")
print(f"Signal : {result.signal}")
print(f"Severity : {result.severity}")
print("-" * 70)
#REAL-TIME LOOP
def run_realtime_monitor(
self,
asset_name: str,
max_cycles: Optional[int] = None,
):
"""
Continuously polls the market data source via SerpApi.
max_cycles=None runs indefinitely.
"""
print("\n" + "=" * 70)
print(f"LAUNCHING ML MARKET MONITOR NODES FOR: {asset_name.upper()}")
print(f"Window: {self.window_size} | Interval: {self.poll_interval}s | Z-Thresh: {self.z_threshold}")
print("=" * 70 + "\n")
cycle_count = 0
try:
while True:
if max_cycles is not None and cycle_count >= max_cycles:
print("[MONITOR] Maximum configured verification cycles reached.")
break
market_price = self.fetch_live_market_price(asset_name)
if market_price is not None:
evaluation = self.evaluate_market_anomaly(market_price)
# Intercept high-signal alerts while respecting cooldown rules
if evaluation.signal != "STABLE" and evaluation.signal != "INITIALIZING":
if self.should_emit_alert():
print(f"\n⚠️ [🚨 ALERT SIGNAL METRIC CAPTURED] {evaluation.signal} Triggered!")
self.display_result(evaluation)
else:
print(f"[COOLDOWN] Anomaly parsed ({evaluation.signal}), throttling alert emit.")
else:
self.display_result(evaluation)
else:
print("[WARNING] Skipped tracking cycle due to data fetch latency drop.")
cycle_count += 1
if max_cycles is not None and cycle_count >= max_cycles:
continue
time.sleep(self.poll_interval)
except KeyboardInterrupt:
print("\n[MONITOR] Graceful execution shutdown triggered via user terminal console.")
def main():
# Production parameter gathering via secure system paths
API_TOKEN = os.getenv("SERPAPI_API_KEY")
if not API_TOKEN:
print("[Abort] System Error: Configure your SERPAPI_API_KEY environment variable.")
print("Example: export SERPAPI_API_KEY='your_key_here'")
sys.exit(1)
# Initialize tracking nodes targeting premium consumer indices
monitor_item = "iPhone 16 Pro"
engine = PriceAlertEngine(
api_key=API_TOKEN,
window_size=12,
poll_interval=15, # Accelerated cycle for local verification testing
z_threshold=3.5,
return_threshold=0.03
)
# Run a controlled 5-cycle loop to populate historical arrays cleanly
engine.run_realtime_monitor(asset_name=monitor_item, max_cycles=5)
if __name__ == "__main__":
main()
create the requirements.txt file and include
google-search-results==2.4.2
requests==2.31.0
numpy==1.26.4
Install SerpApi's Python package:
pip install serpapi
Set your API key:
export SERPAPI_API_KEY="YOUR_ACTUAL_SERPAPI_KEY"
Then run:
python price_alert_engine.py
The example configuration monitors an iPhone 16 Pro with:
window_size=12
poll_interval=15
z_threshold=3.5
return_threshold=0.03
and performs five cycles for local verification.
Example execution flow
A typical run starts by collecting enough observations to initialize the statistical baseline.
During initialization, the detector returns:
Signal : INITIALIZING
Severity : INFO
Once there is enough history, each new observation is evaluated against the existing price window.
A simplified flow looks like:
Observation 1 → INITIALIZING
Observation 2 → INITIALIZING
Observation 3 → INITIALIZING
Observation 4
│
├── Calculate median
├── Calculate MAD
├── Calculate robust Z-score
├── Calculate short-term return
├── Check trend
└── Generate signal
▼
STABLE / PRICE_SPIKE
/PRICE_DROP /
BULLISH_ANOMALY /
BEARISH_ANOMALY
Important engineering lessons
1. Robust statistics are useful for messy data
Real-world retail data is not perfectly clean.
Scraping errors, unusual sellers, temporary discounts, and extreme listings can create observations that should not dominate the baseline.
Median and MAD give us a simple way to reduce the influence of those observations.
2. One metric is rarely enough
The engine combines:
Baseline
+
Dispersion
+
Return
+
Trend
+
Cooldown
Each component answers a different question.
3. Detection and alerting should be separate
A monitoring system should be able to detect repeatedly without necessarily notifying repeatedly.
That separation makes the system much easier to operate.
4. Start interpretable before reaching for complex models
This detector does not require a neural network or a large training dataset.
Every alert can be explained:
Current price
→ baseline
→ MAD
→ robust Z-score
→ return
→ trend
→ final signal
That is valuable when you need to debug false positives.
Where this project can go next
This implementation is intentionally lightweight, but it provides a foundation for a more complete monitoring platform.
Possible extensions include:
SerpApi
│
▼
Price Collector
│
▼
Anomaly Engine
│
├── PostgreSQL / TimescaleDB
├── Redis
├── Prometheus
└── Event Queue
│
▼
Notification Layer
├── Email
├── Telegram
├── Slack
└── Webhook
Other useful improvements would be:
- Per-product thresholds
- Seller-level filtering
- Persistent historical storage
- Structured JSON logging
- Retry and backoff policies
- Multi-market / multi-currency support
- A web dashboard
- More sophisticated change-point detection
- Evaluation against labeled historical anomalies
At that point, the detector becomes more than a script: it becomes a small observability service for retail pricing.
Final thoughts
The interesting part of this project is not the amount of code.
It is the decision to make the detector robust, explainable, and resistant to noisy observations.
By combining a rolling median, MAD, robust Z-scores, percentage returns, trend confirmation, and cooldown-based alerting, we get a practical monitoring pipeline without introducing unnecessary model complexity.
For many real-world monitoring problems, that is a good engineering starting point:
Build the simplest detector that behaves correctly, make its decisions explainable, and only add complexity when the data proves you need it.
Project structure
A minimal project can look like:
price-alert-engine/
├── price_alert_engine.py
├── README.md
└── .env
Keep API credentials out of source control and load them from environment variables.
Source code
The implementation used in this article contains the complete PriceAlertEngine, including market-data retrieval, MAD calculation, robust Z-score computation, anomaly classification, cooldown handling, terminal output, and the real-time polling loop.
get more from my repository:
ssebinacharles
/
quant-bot-command-center
future...
This article was originally published by DEV Community and written by Ssebina CHARLES.
Read original article on DEV Community