DEV Community: Ssebina CHARLES

Building a Real-Time Price Anomaly Detector with Python, SerpApi, and Robust Statistics

Ssebina CHARLES — Sun, 06 Sep 2026 12:32:26 +0000

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:

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:

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...

Building a Lightweight Financial News Sentiment Analyzer with Python and SerpApi

Ssebina CHARLES — Tue, 01 Sep 2026 11:21:08 +0000

Building a Lightweight Financial News Sentiment Analyzer with Python and SerpApi

Tags: python api finance sentimentanalysis beginners

Description:
Learn how to build a simple financial news sentiment analyzer in Python using SerpApi. The project retrieves recent news for assets such as Apple, Tesla, and Bitcoin, then classifies the language as bullish, bearish, or neutral using a lightweight lexicon-based approach.

## Introduction

Financial markets move quickly.

Every day, investors are exposed to thousands of headlines discussing earnings, company announcements, market movements, upgrades, downgrades, risks, and economic events.

The problem is not necessarily finding financial news.

The problem is processing large amounts of news quickly.

In this project, I built a small Python-based financial sentiment engine that retrieves recent financial news and automatically classifies each article as:

🟢 Bullish
🔴 Bearish
🟡 Neutral

The project uses SerpApi to retrieve Google News results and Python to perform the sentiment classification.
if you dont have SerpApi account please visit https://serpapi.com to grap one

The goal is not to build a magical stock-price predictor.

Instead, this project demonstrates how a developer can combine:

Python
   ↓
SerpApi
   ↓
Google News
   ↓
Financial headlines
   ↓
Text processing
   ↓
Sentiment scoring
   ↓
Bullish / Bearish / Neutral

This is a lightweight foundation that can later be extended into a much more advanced financial research system.

What We Are Building

The application accepts an asset such as:

Apple
Tesla
Bitcoin

It then searches for recent financial news related to that asset.

For example:

Apple financial market news

The returned articles are processed and the program extracts:

Title
Snippet
Source

The title and snippet are combined and passed through a simple sentiment engine.

A simplified example might look like:

Headline:
Apple reports record growth in quarterly revenue

Detected terms:
record
growth

Result:
BULLISH 🟢

Another headline might contain:

Apple shares fall after weak earnings outlook

Detected terms:
fall

Result:
BEARISH 🔴

And an article without enough bullish or bearish language might produce:

NEUTRAL 🟡

Project Architecture

The project is intentionally simple.

                  ┌─────────────────────┐
                  │      Python App     │
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │      SerpApi        │
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │    Google News      │
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │  News Headlines     │
                  │  + Snippets        │
                  └──────────┬──────────┘
                             │
                             ▼
                  ┌─────────────────────┐
                  │ Sentiment Engine    │
                  └──────────┬──────────┘
                             │
                ┌────────────┼────────────┐
                ▼            ▼            ▼
             Bullish      Neutral      Bearish
               🟢            🟡            🔴

The current implementation is contained in a single Python file, `setiment_engine.py`. It defines a `FinancialSentimentEngine` class, bullish and bearish token sets, a news retrieval method, a sentiment analyzer, and a pipeline runner.

Requirements

You need:

Python 3
A SerpApi account/API key
Internet access
A terminal
The serpapi Python package

The official SerpApi Python package can be installed with:

pip install serpapi

The current SerpApi Python client is based around serpapi.Client, which is the approach used in this project.

Step 1: Create the Project

Create a project directory:

mkdir financial-sentiment
cd financial-sentiment

Create a virtual environment:

python -m venv .venv

Activate it on Windows:

.venv\Scripts\activate

You should see something similar to:

(.venv) C:\Users\YourName\financial-sentiment>

Step 2: Install SerpApi

Upgrade pip:

python -m pip install --upgrade pip

Then install the official SerpApi package:

python -m pip install serpapi

Verify the installation:

python -m pip show serpapi

The important detail here is that the modern client uses:

import serpapi

client = serpapi.Client(api_key="...")

rather than the older:

from serpapi import GoogleSearch

This difference is important because using the wrong interface can result in:

ImportError: cannot import name 'GoogleSearch' from 'serpapi'

The official SerpApi Python repository identifies serpapi.Client as the primary interface and recommends the serpapi package.

Step 3: Configure the API Key

Do not hard-code your API key into your Python source.

Instead, store it in an environment variable.

On Windows CMD:

set SERPAPI_API_KEY=YOUR_API_KEY

For a persistent Windows environment variable:

setx SERPAPI_API_KEY "YOUR_API_KEY"

After using setx, open a new terminal.

You can check whether the variable exists without printing the secret:

if defined SERPAPI_API_KEY (echo API key is configured) else (echo API key is missing)

A good result is:

API key is configured

Environment variables are also the approach recommended by the official SerpApi Python package documentation for keeping API keys separate from application code.

Step 4: The Python Application

Here is the complete implementation.

import os
import re
import sys
import time

import serpapi


class FinancialSentimentEngine:
    """Fetch financial news and classify its sentiment."""

    BULLISH_TOKENS = {
        "growth",
        "surge",
        "gain",
        "profit",
        "bullish",
        "higher",
        "record",
        "success",
        "buy",
        "upgrade",
    }

    BEARISH_TOKENS = {
        "drop",
        "fall",
        "loss",
        "crash",
        "bearish",
        "lower",
        "risk",
        "decline",
        "sell",
        "downgrade",
    }

    def __init__(self, api_key: str):
        if not api_key:
            raise ValueError("Missing SERPAPI_API_KEY.")

        self.client = serpapi.Client(api_key=api_key)

    def fetch_market_news(self, ticker: str) -> list:
        """Fetch recent financial news using SerpApi Google News."""

        print(
            f"[Engine] Querying financial news for ticker: {ticker}..."
        )

        search_parameters = {
            "engine": "google",
            "q": f"{ticker} financial market news",
            "tbm": "nws",
            "tbs": "qdr:w",
        }

        try:
            dictionary_results = self.client.search(search_parameters)

            if "error" in dictionary_results:
                print(
                    f"[Error] SerpApi returned an error: "
                    f"{dictionary_results['error']}"
                )
                return []

            return dictionary_results.get("news_results", [])

        except Exception as exc:
            print(f"[Error] Network query failed: {exc}")
            return []

    @classmethod
    def analyze_text_sentiment(cls, text: str) -> str:
        """Perform lightweight lexicon-based sentiment classification."""

        normalized_text = text.lower()

        bull_score = sum(
            1
            for token in cls.BULLISH_TOKENS
            if re.search(
                rf"\b{re.escape(token)}\b",
                normalized_text
            )
        )

        bear_score = sum(
            1
            for token in cls.BEARISH_TOKENS
            if re.search(
                rf"\b{re.escape(token)}\b",
                normalized_text
            )
        )

        if bull_score > bear_score:
            return "BULLISH 🟢"

        if bear_score > bull_score:
            return "BEARISH 🔴"

        return "NEUTRAL 🟡"

    def run_pipeline(self, ticker: str) -> None:
        """Fetch news and classify the first five articles."""

        news_articles = self.fetch_market_news(ticker)

        if not news_articles:
            print(
                f"[Warning] No news found for {ticker} "
                "within the selected time window."
            )
            return

        print(f"\n=== SENTIMENT REPORT: {ticker.upper()} ===")

        for index, article in enumerate(
            news_articles[:5],
            start=1
        ):
            title = article.get("title", "No title")
            snippet = article.get("snippet", "")

            source = article.get(
                "source",
                "Unknown source"
            )

            if isinstance(source, dict):
                source = source.get(
                    "name",
                    "Unknown source"
                )

            sentiment = self.analyze_text_sentiment(
                f"{title} {snippet}"
            )

            print(f"\n[{index}] Source: {source}")
            print(f"    Headline: {title}")
            print(f"    Sentiment: {sentiment}")


def main():
    api_key = os.getenv("SERPAPI_API_KEY")

    if not api_key:
        print(
            "[Abort] Set the SERPAPI_API_KEY "
            "environment variable first."
        )
        sys.exit(1)

    engine = FinancialSentimentEngine(api_key)

    assets = [
        "Apple",
        "Tesla",
        "Bitcoin",
    ]

    for index, asset in enumerate(assets):
        engine.run_pipeline(asset)

        print("-" * 60)

        if index < len(assets) - 1:
            time.sleep(1)


if __name__ == "__main__":
    main()

This implementation follows the structure of the uploaded source while correcting the SerpApi client usage and indentation problems that caused the earlier failure.

Step 5: Understanding the Sentiment Engine

The heart of the project is the sentiment classifier.

We define two groups of words.

Bullish words

BULLISH_TOKENS = {
    "growth",
    "surge",
    "gain",
    "profit",
    "bullish",
    "higher",
    "record",
    "success",
    "buy",
    "upgrade",
}

These terms generally represent positive market language.

Bearish words

BEARISH_TOKENS = {
    "drop",
    "fall",
    "loss",
    "crash",
    "bearish",
    "lower",
    "risk",
    "decline",
    "sell",
    "downgrade",
}

These represent negative market language.

The project then normalizes the article text:

normalized_text = text.lower()

This prevents capitalization from affecting the matching.

For example:

Growth
growth
GROWTH

all become:

growth

Why Use Regular Expressions?

Instead of checking whether a token simply exists somewhere inside the text, the implementation uses word boundaries:

re.search(
    rf"\b{re.escape(token)}\b",
    normalized_text
)

This helps reduce accidental partial matches.

For example, we want:

risk

to match the word:

risk

rather than accidentally matching unrelated words that merely contain the same character sequence.

Calculating the Scores

The bullish score is calculated with:

bull_score = sum(
    1
    for token in cls.BULLISH_TOKENS
    if re.search(
        rf"\b{re.escape(token)}\b",
        normalized_text
    )
)

The bearish score works the same way:

bear_score = sum(
    1
    for token in cls.BEARISH_TOKENS
    if re.search(
        rf"\b{re.escape(token)}\b",
        normalized_text
    )
)

The final decision is simple:

bullish score > bearish score
        ↓
     BULLISH

bearish score > bullish score
        ↓
     BEARISH

scores are equal
        ↓
     NEUTRAL

The uploaded implementation follows exactly this three-way decision structure.

Step 6: Retrieving Financial News

The application creates a SerpApi search request:

search_parameters = {
    "engine": "google",
    "q": f"{ticker} financial market news",
    "tbm": "nws",
    "tbs": "qdr:w",
}

The important pieces are:

`engine`

"engine": "google"

This specifies the Google search engine interface.

`q`

"q": f"{ticker} financial market news"

For:

Apple

the query becomes:

Apple financial market news

`tbm`

"tbm": "nws"

This is used to target Google News results.

`tbs`

"tbs": "qdr:w"

This asks for a recent time window of approximately one week.

SerpApi's client accepts Google search parameters through client.search(...), with results returned in a dictionary-like object.

Step 7: Extracting Articles

The application retrieves:

news_articles = self.fetch_market_news(ticker)

Then it processes only the first five results:

for index, article in enumerate(
    news_articles[:5],
    start=1
):

For every article, it extracts:

title = article.get("title", "No title")
snippet = article.get("snippet", "")

It also handles the source field because the returned source data can be represented as a dictionary:

source = article.get(
    "source",
    "Unknown source"
)

if isinstance(source, dict):
    source = source.get(
        "name",
        "Unknown source"
    )

Finally, the title and snippet are combined:

sentiment = self.analyze_text_sentiment(
    f"{title} {snippet}"
)

This gives the sentiment engine more context than the title alone.

Step 8: Running the Pipeline

The main() function reads the API key:

api_key = os.getenv("SERPAPI_API_KEY")

Then it creates the engine:

engine = FinancialSentimentEngine(api_key)

The demonstration analyzes:

assets = [
    "Apple",
    "Tesla",
    "Bitcoin",
]

The pipeline runs once for each asset.

Step 9: Run the Program

Save the program as:

setiment_engine.py

Then run:

python setiment_engine.py

You should see output similar to:

[Engine] Querying financial news for ticker: Apple...

=== SENTIMENT REPORT: APPLE ===

[1] Source: Example News
    Headline: Apple reports strong revenue growth
    Sentiment: BULLISH 🟢

[2] Source: Example News
    Headline: Apple faces concerns about lower demand
    Sentiment: BEARISH 🔴

------------------------------------------------------------
[Engine] Querying financial news for ticker: Tesla...

=== SENTIMENT REPORT: TESLA ===

[1] Source: Example News
    Headline: Tesla announces higher production
    Sentiment: BULLISH 🟢

The exact articles and classifications will change because the program retrieves current search results at runtime.

What Makes This Project Useful?

At first glance, counting words might look too simple.

And it is simple.

But simple systems can be useful as a starting point.

This project demonstrates several important software engineering concepts:

API integration

The application communicates with an external service and consumes structured search results.

Environment-based configuration

The API credential is separated from the source code.

Object-oriented design

The financial engine is encapsulated in:

class FinancialSentimentEngine:

Error handling

The application catches failed requests and handles missing search results.

Text processing

The application normalizes text and analyzes lexical features.

Pipeline architecture

The complete process is divided into stages:

Input
 ↓
Search
 ↓
Extraction
 ↓
Normalization
 ↓
Scoring
 ↓
Classification
 ↓
Output

Important Limitation: This Is Not a Trading Signal

This is probably the most important part of the project.

The program is not predicting the future price of a stock or cryptocurrency.

It is also not a professional quantitative trading model.

It does not account for:

Market structure
Price action
Trading volume
Macroeconomic data
Earnings estimates
Technical indicators
Institutional positioning
Options activity
News credibility
Context
Negation
Sarcasm
Historical correlations

For example, consider:

Apple avoids a major revenue decline despite weaker demand.

A simple lexical system may see:

decline
weaker

and classify the article negatively, even though the overall meaning could be more nuanced.

That is one of the fundamental weaknesses of keyword-based sentiment analysis.

The system should therefore be viewed as a news-research utility, not as an automated investment decision maker.

How I Would Improve This Project

This project provides a foundation for something much more advanced.

The next version could introduce a proper NLP model instead of manually defined keywords.

For example:

                  Financial News
                        │
                        ▼
                Text Cleaning
                        │
                        ▼
               NLP Model / LLM
                        │
                        ▼
              Sentiment Score
                        │
            ┌───────────┼───────────┐
            ▼           ▼           ▼
         Bullish      Neutral     Bearish

The output could become numerical:

{
  "asset": "Apple",
  "sentiment": "bullish",
  "score": 0.82
}

instead of simply:

BULLISH 🟢

Possible Future Features

The project could eventually become a complete financial research platform.

For example:

                    Financial Research Platform
                              │
          ┌───────────────────┼───────────────────┐
          ▼                   ▼                   ▼
       News API           Market Data          Social Data
          │                   │                   │
          └───────────────────┼───────────────────┘
                              ▼
                       NLP / AI Analysis
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
          Sentiment        Trends          Events
              │               │               │
              └───────────────┼───────────────┘
                              ▼
                       Research Dashboard

Possible additions include:

Historical sentiment storage
PostgreSQL
REST APIs
FastAPI
Redis caching
Background workers
Market price APIs
Technical indicators
Sentiment charts
News deduplication
Source credibility scoring
LLM-based financial summarization
Portfolio monitoring
Web dashboards
Alerts

That would transform this from a small command-line experiment into a much more complete financial intelligence platform.

Security Notes

Never put your real API key inside:

api_key = "my-real-secret-key"

Do not commit API keys to GitHub.

Do not put them in screenshots.

Do not publish them in DEV.to articles.

Use:

os.getenv("SERPAPI_API_KEY")

and configure the key through your environment.

If a key is accidentally published, treat it as compromised and rotate it.

A Note About the Name

The file in this project is currently called:

setiment_engine.py

The intended spelling is probably:

sentiment_engine.py

The typo does not affect Python functionality, but I recommend renaming it before publishing the repository:

ren setiment_engine.py sentiment_engine.py

Then run:

python sentiment_engine.py

That will make the public project look more polished.

Project Structure

For the first version, the project can remain small:

financial-sentiment/
│
├── .venv/
│
├── sentiment_engine.py
│
├── .gitignore
│
└── README.md

Your .gitignore should include:

.venv/
.env
__pycache__/
*.pyc

Final Result

With only Python and SerpApi, we now have a working pipeline capable of:

Asset
  ↓
Financial News Search
  ↓
Recent Articles
  ↓
Headline + Snippet
  ↓
Lexicon Analysis
  ↓
Bullish / Bearish / Neutral

It is small, understandable, and easy to extend.

More importantly, it demonstrates an important engineering principle:

You do not need to begin with an enormous AI system to build something useful.

Start with a clear pipeline.

Then improve each stage.

Conclusion

This project started as a simple question:

Can we automatically process recent financial news and get a quick indication of its overall tone?

Using Python and SerpApi, the answer is yes.

The current implementation is deliberately lightweight. It uses a manually defined financial vocabulary and compares bullish and bearish word counts.

That makes it easy to understand and easy to modify.

But it is only the beginning.

A future version could combine real market data, historical sentiment, machine learning, NLP, and large language models to build a far more sophisticated financial research system.

For now, this project provides a clean starting point for developers who want to learn how to connect Python to an external API and turn raw news data into a simple analytical output.

**The next step is not making the code bigger.

The next step is making the analysis smarter.**

Disclaimer

This project is for educational and research purposes only. The sentiment output is not financial advice and should not be used as the sole basis for investment or trading decisions.

The importance of the research

Am in a developing phase of building an autonomous AI trading quantum system for more visit https://github.com/ssebinacharles/quant-bot-command-center

How to Build a Resilient Market Data Aggregator in Python using SerpApi

Ssebina CHARLES — Thu, 27 Aug 2026 10:01:58 +0000

When building automated backend systems like trading bots, algorithmic platforms, or financial scrapers, connection resilience is critical. A simple remote network hiccup or temporary API gateway timeout shouldn't completely crash your entire runtime script execution pipeline.

In this tutorial, we will build a production-grade Python command-line utility that extracts real-time stock, market index, and cryptocurrency metrics via SerpApi's Google Finance engine. We will explicitly implement custom exception routing and an exponential backoff retry loop to handle unstable network connections safely.

Prerequisites & Setup

To follow along with this implementation, ensure you have Python 3.x installed on your local operating system.

First, open your terminal environment and install the required external HTTP network connection module:

pip install requests

Next, ensure you register for a free developer access key by visiting https://serpapi.com/dashboard. To protect your private authentication credentials from being hardcoded inside your public script files, inject your key directly as a local session environment variable in your active terminal terminal panel:

Windows Command Prompt (cmd):

set SERPAPI_KEY=your_private_api_key_here

Mac / Linux Terminal:

export SERPAPI_KEY="your_private_api_key_here"

The Resilient Code Engine

Create a new file in your local workspace using Notepad++ or your preferred code editor, name it exactly serpapi_market_data.py, and paste the following fault-tolerant implementation structure:

import os
import sys
import time
from typing import Optional
import requests

SERPAPI_ENDPOINT = "https://serpapi.com"
DEFAULT_TIMEOUT = 10  # seconds
MAX_RETRIES = 3
RETRY_BACKOFF = 2  # seconds

class SerpApiError(Exception):
    """Custom exception routing configurations for SerpApi specific runtime errors."""

def get_api_key() -> str:
    api_key = os.environ.get("SERPAPI_KEY")
    if not api_key:
        raise SerpApiError("No API key found. Ensure the SERPAPI_KEY environment variable is configured.")
    return api_key

def fetch_quote(ticker: str, api_key: str) -> dict:
    params = {
        "engine": "google_finance",
        "q": ticker,
        "api_key": api_key,
    }
    last_error: Optional[Exception] = None

    for attempt in range(1, MAX_RETRIES + 1):
        try:
            response = requests.get(params=params, url=SERPAPI_ENDPOINT, timeout=DEFAULT_TIMEOUT)
            response.raise_for_status()
            data = response.json()

            if "error" in data:
                raise SerpApiError(f"SerpApi dashboard error for {ticker}: {data['error']}")
            return data

        except (requests.RequestException, SerpApiError) as exc:
            last_error = exc
            if attempt < MAX_RETRIES:
                sleep_time = RETRY_BACKOFF * attempt
                print(f"[warn] {ticker}: Attempt {attempt} failed. Retrying execution in {sleep_time}s...")
                time.sleep(sleep_time)

    raise SerpApiError(f"Failed to fetch tracking data for {ticker} after {MAX_RETRIES} attempts. Engine error: {last_error}")

def parse_quote(data: dict, ticker: str) -> dict:
    summary = data.get("summary", {})
    if not summary:
        raise SerpApiError(f"No summary data structures returned for {ticker}. Verify parameter definitions.")
    return {
        "ticker": ticker,
        "title": summary.get("title"),
        "price": summary.get("extracted_price") or summary.get("price"),
        "currency": summary.get("currency"),
        "change": summary.get("price_movement", {}).get("value"),
        "change_percent": summary.get("price_movement", {}).get("percentage"),
        "movement": summary.get("price_movement", {}).get("movement"),
    }

def main():
    tickers = sys.argv[1:] or ["AAPL:NASDAQ", "GOOGL:NASDAQ", "BTC-USD"]
    try:
        api_key = get_api_key()
    except SerpApiError as fatal_err:
        print(f"[fatal] {fatal_err}")
        sys.exit(1)

    print(f"\n{'Ticker':<15}{'Price':<12}{'Change':<12}{'% Change':<12}{'Movement':<10}")
    print("-" * 61)

    for ticker in tickers:
        try:
            raw = fetch_quote(ticker, api_key)
            r = parse_quote(raw, ticker)
            print(f"{r['ticker']:<15}{str(r['price']):<12}{str(r['change']):<12}{str(r['change_percent']):<12}{str(r['movement']):<10}")
        except SerpApiError as exc:
            print(f"{ticker:<15}ERROR BLOCK: {exc}")

if __name__ == "__main__":
    main()

Execution & Grid Verification

Execute the operational script engine directly within your terminal window while supplying any target asset trackers as arguments:

python serpapi_market_data.py AAPL:NASDAQ GOOGL:NASDAQ BTC-USD

The application layer will automatically process the commands, query the server endpoints, run the mathematical exponential backoff algorithm if an unexpected network timeout happens, and print out a clean, structured asset data grid directly into your console pipeline.

Check out my working software tools and open-source backend layouts on my public portfolio link: (https://github.com/ssebinacharles/quant-bot-command-center).

DEV Community: Ssebina CHARLES

Building a Real-Time Price Anomaly Detector with Python, SerpApi, and Robust Statistics

What we're building

Why not just use standard deviation?

1. Building a rolling median baseline

2. Median Absolute Deviation (MAD)

3. Turning MAD into a robust Z-score

Why 0.6745?

4. Detecting short-term price movement

5. Confirming the trend

The implementation assigns higher severity to confirmed rising or falling anomalies.

6. Combining the signals

Getting real retail prices with SerpApi

The result object

Avoiding alert spam

Running the monitor

Complete minimal setup

Example execution flow

Important engineering lessons

1. Robust statistics are useful for messy data

2. One metric is rarely enough

3. Detection and alerting should be separate

4. Start interpretable before reaching for complex models

Where this project can go next

Final thoughts

Project structure

Source code

ssebinacharles / quant-bot-command-center

future...

Building a Lightweight Financial News Sentiment Analyzer with Python and SerpApi

Building a Lightweight Financial News Sentiment Analyzer with Python and SerpApi

What We Are Building

Project Architecture

The current implementation is contained in a single Python file, setiment_engine.py. It defines a FinancialSentimentEngine class, bullish and bearish token sets, a news retrieval method, a sentiment analyzer, and a pipeline runner.

Requirements

Step 1: Create the Project

Step 2: Install SerpApi

Step 3: Configure the API Key

Step 4: The Python Application

This implementation follows the structure of the uploaded source while correcting the SerpApi client usage and indentation problems that caused the earlier failure.

Step 5: Understanding the Sentiment Engine

Bullish words

Bearish words

Why Use Regular Expressions?

Calculating the Scores

Step 6: Retrieving Financial News

engine

q

tbm

tbs

Step 7: Extracting Articles

Step 8: Running the Pipeline

Step 9: Run the Program

What Makes This Project Useful?

API integration

Environment-based configuration

Object-oriented design

Error handling

Text processing

Pipeline architecture

Important Limitation: This Is Not a Trading Signal

How I Would Improve This Project

Possible Future Features

Security Notes

A Note About the Name

Project Structure

Final Result

Conclusion

Disclaimer

The importance of the research

How to Build a Resilient Market Data Aggregator in Python using SerpApi

Prerequisites & Setup

The Resilient Code Engine

Execution & Grid Verification

The current implementation is contained in a single Python file, `setiment_engine.py`. It defines a `FinancialSentimentEngine` class, bullish and bearish token sets, a news retrieval method, a sentiment analyzer, and a pipeline runner.

`engine`

`q`

`tbm`

`tbs`