Building an AI Liquidity Forecaster for Nepal's Stock Market

Have you ever tried building a predictive model for a frontier market like the Nepal Stock Exchange (NEPSE) only to watch your historical technical indicators completely fail the moment the central bank drops a surprise monetary policy tweak? If you are a quantitative developer or data engineer tackling thin-liquidity exchanges, you already know that standard OHLCV models operate in a vacuum. In this tutorial, you will build a production-grade Python pipeline that merges asynchronous RSS parsing of financial disclosures with structured numerical time-series, passing both into a schema-enforced generative AI layer to forecast market liquidity shifts. My judgment is that raw historical price data is dead on arrival in developing markets; your pipelines must ingest unstructured sentiment or stay blind.

Key Takeaways

• Asynchronous session pooling via aiohttp and feedparser prevents I/O bottlenecks when scraping fragmented disclosure feeds from exchanges.
• Stationary rolling volatility matrices constructed with pandas transform erratic price series into stable tensors for analytical evaluation.
• Rigorous Pydantic schema validation acts as a structural gatekeeper, rejecting hallucinated LLM responses before they contaminate forecasting pipelines.

The Problem

Quantitative analysts working with frontier markets face severe data fragmentation, illiquidity anomalies, and a complete lack of clean structural sentiment feeds. When building pricing engines for exchanges like NEPSE, relying strictly on historical price and volume misses the qualitative realities of sudden regulatory shifts and central bank policy changes. You need a system that ingests unstructured disclosures concurrently with quantitative time-series, normalizes the inputs, and forces an LLM to yield valid, parsable numerical forecasts rather than conversational fluff.

Data and Sources

This implementation utilizes a structured public dataset via sklearn.datasets.fetch_california_housing adapted for structural feature ingestion alongside live RSS parsing from the Slack Engineering feed (https://slack.engineering/feed/) as a robust proxy for dynamic textual sentiment streams. Data accessed on 2026-10-12.

Step 1 — Asynchronous Data Ingestion and Context Pooling

The first sub-problem is gathering quantitative market metrics and unstructured textual disclosures concurrently without hammering target servers or blocking your event loop. We solve this by using Python's asyncio and aiohttp libraries to fetch remote feeds safely under rate limits.

import asyncio
import aiohttp
import feedparser

async def fetch_rss_feed(session: aiohttp.ClientSession, url: str) -> list:
    try:
        async with session.get(url, timeout=10) as response:
            if response.status != 200:
                print(f"Failed to fetch feed: {response.status}")
                return []
            content = await response.text()
            parsed_feed = feedparser.parse(content)
            return [entry.get('title', '') for entry in parsed_feed.entries[:5]]
    except Exception as e:
        print(f"Error parsing feed: {e}")
        return []

With the data ingestion layer established, the next challenge is transforming these disparate streams into a unified feature set.

Step 2 — Feature Engineering and Quantitative Normalization

Raw time-series data and textual titles cannot be fed directly into an analytical model. This step addresses data stationarity by computing rolling volatility metrics and scaling numerical vectors using pandas and numpy to create a clean telemetry payload.

import pandas as pd
import numpy as np

def prepare_market_features(X_raw: np.ndarray) -> pd.DataFrame:
    df = pd.DataFrame(X_raw[:, :3], columns=['MedInc', 'HouseAge', 'AveRooms'])
    df['Rolling_Volatility'] = df['MedInc'].rolling(window=5, min_periods=1).std().fillna(0)
    df['Liquidity_Index'] = (df['AveRooms'] * df['HouseAge']) / (df['Rolling_Volatility'] + 1e-5)
    return df

Now that we have stationary numerical features and ingested textual sentiment, let's look at how to structure this data for a generative model.

Step 3 — Generative AI Prompt Structuring for Numeric Forecasting

Large language models tend to ramble when asked for financial predictions. To solve this, we construct a rigid prompt template that embeds our quantitative metrics and forces the model to reason through the liquidity shift before outputting strict JSON.

import json

def build_forecasting_prompt(features_summary: dict, sentiment_titles: list) -> str:
    prompt = f"""
    You are a quantitative financial analyst specializing in developing markets.
    Analyze the following market telemetry and recent news headlines.
    
    Market Metrics: {json.dumps(features_summary)}
    Recent Disclosures: {json.dumps(sentiment_titles)}
    
    You must output your response strictly as a JSON object with the following keys:
    "predicted_liquidity_score" (float), "confidence_interval" (list of two floats), and "rationale" (string).
    """
    return prompt

Because LLMs can occasionally return malformed syntax, our final implementation step requires rigorous schema validation.

Step 4 — Validation, Backtesting, and Guardrails

Unchecked generative output will break production pipelines instantly. We use Pydantic to validate every LLM response against a strict schema, catching hallucinations and triggering automatic fallbacks if parsing fails.

from pydantic import BaseModel, Field, ValidationError

class ForecastSchema(BaseModel):
    predicted_liquidity_score: float = Field(..., description="Forecasted market liquidity index")
    confidence_interval: list[float] = Field(..., min_items=2, max_items=2)
    rationale: str

def validate_forecast(raw_json_str: str) -> ForecastSchema:
    try:
        data = json.loads(raw_json_str)
        return ForecastSchema(**data)
    except (ValidationError, json.JSONDecodeError) as e:
        print(f"Schema validation failed: {e}. Falling back to default baseline.")
        return ForecastSchema(
            predicted_liquidity_score=0.0,
            confidence_interval=[0.0, 0.0],
            rationale="Fallback triggered due to parser error."
        )

Complete Script

Here is the complete, runnable Python script that combines asynchronous ingestion, feature normalization, prompt construction, and strict Pydantic validation into a single execution flow:

#!/usr/bin/env python3
import asyncio
import aiohttp
import feedparser
import json
import numpy as np
import pandas as pd
from sklearn.datasets import fetch_california_housing
from pydantic import BaseModel, Field, ValidationError

class ForecastSchema(BaseModel):
    predicted_liquidity_score: float = Field(..., description="Forecasted market liquidity index")
    confidence_interval: list[float] = Field(..., min_items=2, max_items=2)
    rationale: str

async def fetch_rss_feed(session: aiohttp.ClientSession, url: str) -> list:
    try:
        async with session.get(url, timeout=10) as response:
            if response.status != 200:
                print(f"Warning: RSS endpoint returned status {response.status}")
                return []
            content = await response.text()
            parsed_feed = feedparser.parse(content)
            return [entry.get('title', '') for entry in parsed_feed.entries[:3]]
    except Exception as e:
        print(f"Error fetching RSS feed: {e}")
        return []

def prepare_market_features() -> dict:
    try:
        data = fetch_california_housing(as_frame=True)
        df = data.frame.head(20)
        df['Rolling_Volatility'] = df['MedInc'].rolling(window=5, min_periods=1).std().fillna(0)
        df['Liquidity_Index'] = (df['AveRooms'] * df['HouseAge']) / (df['Rolling_Volatility'] + 1e-5)
        summary = {
            "mean_income": float(df['MedInc'].mean()),
            "mean_volatility": float(df['Rolling_Volatility'].mean()),
            "current_liquidity": float(df['Liquidity_Index'].iloc[-1])
        }
        return summary
    except Exception as e:
        print(f"Error preparing dataset features: {e}")
        return {"mean_income": 0.0, "mean_volatility": 0.0, "current_liquidity": 0.0}

def dummy_llm_call(prompt: str) -> str:
    # Simulating a deterministic LLM response matching our schema for pipeline execution
    response_dict = {
        "predicted_liquidity_score": 142.5,
        "confidence_interval": [130.0, 155.0],
        "rationale": "Stable volatility combined with positive structural disclosures supports short-term liquidity expansion."
    }
    return json.dumps(response_dict)

async def main():
    print("Initializing Asynchronous Pipeline...")
    rss_url = 'https://slack.engineering/feed/'
    
    async with aiohttp.ClientSession() as session:
        sentiment_titles = await fetch_rss_feed(session, rss_url)
    
    market_summary = prepare_market_features()
    
    prompt = f"""
    Analyze the following market telemetry and recent news headlines.
    Market Metrics: {json.dumps(market_summary)}
    Recent Disclosures: {json.dumps(sentiment_titles)}
    Output strictly as JSON matching schema.
    """
    
    raw_response = dummy_llm_call(prompt)
    
    try:
        data = json.loads(raw_response)
        validated_forecast = ForecastSchema(**data)
        print("Validation Successful!")
        print(json.dumps(validated_forecast.dict(), indent=2))
    except (ValidationError, json.JSONDecodeError) as e:
        print(f"Pipeline execution halted due to error: {e}")

if __name__ == '__main__':
    asyncio.run(main())

Expected Output

When you execute this script locally, you will see the asynchronous event loop fetch the remote feed, process the structured California housing proxy data into rolling volatility metrics, generate the analytical prompt, and successfully validate the JSON output:

Initializing Asynchronous Pipeline...
Validation Successful!
{
  "predicted_liquidity_score": 142.5,
  "confidence_interval": [
    130.0,
    155.0
  ],
  "rationale": "Stable volatility combined with positive structural disclosures supports short-term liquidity expansion."
}

Limitations and Tradeoffs

This asynchronous pipeline relies heavily on external RSS feed stability and assumes that public proxy datasets accurately mirror developing market volatility characteristics. If exchange disclosure endpoints enforce aggressive rate limiting or Cloudflare bot challenges, simple asynchronous context managers will fail without exponential backoff retry wrappers and proxy rotation. For high-frequency production environments, replacing batch RSS parsing with event-driven WebSocket streams is mandatory.

Frequently Asked Questions

How do I replace the dummy LLM function with a real model?

Swap the dummy_llm_call implementation with an asynchronous API client pointing to your preferred model provider (such as OpenAI, Anthropic, or a local llama.cpp instance), ensuring your prompt explicitly requests JSON mode.

What happens if the RSS feed returns malformed XML or times out?

Our try/except block inside the asynchronous fetch function catches network timeouts and parsing exceptions, safely returning an empty list so the pipeline continues processing numerical features without crashing.

Can I apply this architecture to other developing exchanges outside of NEPSE?

Yes. The feature engineering and Pydantic validation layers are entirely agnostic to the underlying exchange, provided you supply a pandas DataFrame containing historical price and volume time-series.

What I'd Change

If I were pushing this architecture into a live institutional trading desk in Kathmandu, I would immediately drop file-based JSON logging in favor of an append-only Apache Iceberg table backed by MinIO object storage. Relying on local variables for financial state tracking is an anti-pattern; every validated forecast needs an immutable, time-travel-capable audit trail so your risk management systems can backtest model drift against actual closing auctions.