Skip to content
Home
finance-nepal
nepse

From Todos to Trades: Building a NEPSE Moving‑Average Screener in 30 Minutes

Mausam

When I needed a quick proof‑of‑concept for a NEPSE technical‑analysis tool, the only data I had on hand was a public placeholder API. It sounded absurd – “how can a list of todo items become a trading signal?” – but the exercise forced me to design a pipeline that fetches, caches, normalises, computes indicators, detects crossovers, and persists results, all while handling the edge cases you hit in a real CI/CD job. If you’re a quant, data engineer, or fintech hobbyist who wants a lightweight, reproducible screener you can drop into Airflow or a GitHub Action, this walk‑through shows exactly how to do it.

Key Takeaways

  • Cache remote JSON once per run to guarantee idempotent results and avoid rate‑limits.
  • Transform arbitrary JSON into a time‑ordered price series without a dedicated market feed.
  • Compute overlapping short‑ and long‑window moving averages with pandas, then flag golden/death crossovers in a single pass.
  • Persist signals as a CSV that downstream trading bots can ingest, and optionally visualise them with a saved PNG.
  • Wrap network and I/O operations in explicit try/except blocks to keep the pipeline robust in production.

The Problem

Financial analysts at Nepse often need a “plug‑and‑play” screener that runs nightly, spits out a CSV of buy/sell flags, and can be version‑controlled. Existing tutorials either assume you have a live market feed or they hide the plumbing behind heavyweight frameworks. I wanted a minimal, auditable script that could be run from a cron job, a Docker container, or an Airflow task without pulling in a full‑blown data lake.

Data and Sources

We use the public JSONPlaceholder Todos endpoint as a stand‑in for a price feed. Each todo’s id will act as a timestamp (seconds since epoch) and the userId will be interpreted as a price. The endpoint is stable and free for experimentation. Data accessed on 2026-10-31.

Step 1 — Fetch & Cache the API

The first sub‑problem is ensuring we don’t hammer the placeholder service and that our script is deterministic across runs. We download the JSON once, write it to a local cache file, and reuse it if it already exists.

import json, os, requests, hashlib

CACHE_PATH = "todos_cache.json"

def fetch_todos():
    if os.path.exists(CACHE_PATH):
        with open(CACHE_PATH, "r") as f:
            return json.load(f)

    try:
        resp = requests.get("https://jsonplaceholder.typicode.com/todos", timeout=5)
        resp.raise_for_status()
        data = resp.json()
        with open(CACHE_PATH, "w") as f:
            json.dump(data, f)
        return data
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to fetch todos: {e}")

We first look for todos_cache.json. If it’s missing we perform a GET with a short timeout, raise for HTTP errors, and dump the payload locally. A network failure bubbles up as a clear RuntimeError, which the orchestrator can catch and retry.

Step 2 — Normalize to a Price Series

Our raw todos have no time dimension, so we coerce id into a monotonic timestamp (seconds since the epoch) and userId into a synthetic price. The result is a Pandas DataFrame sorted by timestamp.

import pandas as pd
from datetime import datetime, timedelta

def todos_to_price_series(todos):
    # Map each todo to a timestamp spaced one minute apart
    base = datetime(2023, 1, 1)
    rows = []
    for todo in todos:
        ts = base + timedelta(minutes=todo["id"])
        price = float(todo["userId"])  # simple price proxy
        rows.append({"timestamp": ts, "price": price})
    df = pd.DataFrame(rows).set_index("timestamp")
    return df.sort_index()

Using a fixed start date guarantees reproducibility. The function returns a time‑indexed DataFrame ready for rolling calculations.

Step 3 — Compute Short & Long Moving Averages

Technical analysts typically compare a fast (e.g., 5‑period) MA with a slower (e.g., 20‑period) MA. Pandas’ rolling does the heavy lifting in O(N) time.

def add_moving_averages(df, short=5, long=20):
    df["ma_short"] = df["price"].rolling(window=short, min_periods=1).mean()
    df["ma_long"]  = df["price"].rolling(window=long,  min_periods=1).mean()
    return df

We keep min_periods=1 so the early rows still have a value, which simplifies downstream logic.

Step 4 — Detect Crossovers & Label Signals

A “golden cross” (short MA crossing above long MA) is a buy signal; the opposite is a sell signal. We detect the sign change of the difference between the two MAs.

def label_signals(df):
    df["diff"] = df["ma_short"] - df["ma_long"]
    df["signal"] = 0  # neutral by default
    # Shift diff to compare current vs previous
    df["prev_diff"] = df["diff"].shift(1).fillna(0)

    buy_mask  = (df["diff"] > 0) & (df["prev_diff"] <= 0)
    sell_mask = (df["diff"] < 0) & (df["prev_diff"] >= 0)

    df.loc[buy_mask,  "signal"] = 1   # 1 = buy
    df.loc[sell_mask, "signal"] = -1  # -1 = sell
    return df.drop(columns=["diff", "prev_diff"])

The logic is vectorised: we compute the difference, shift it, and flag rows where the sign flips. The resulting signal column is ready for downstream consumption.

Step 5 — Persist & Expose Results

Finally we write a CSV that downstream bots can read, and we optionally generate a PNG chart for visual sanity checks.

import matplotlib.pyplot as plt

def persist_results(df, csv_path="signals.csv", img_path="ma_chart.png"):
    try:
        df.to_csv(csv_path, index_label="timestamp")
    except OSError as e:
        raise RuntimeError(f"Unable to write CSV: {e}")

    # Plot only if a graphical backend is available
    try:
        plt.figure(figsize=(10, 4))
        df["price"].plot(label="Price")
        df["ma_short"].plot(label="MA 5")
        df["ma_long"].plot(label="MA 20")
        plt.legend()
        plt.title("Synthetic Price + Moving Averages")
        plt.savefig(img_path)
        plt.close()
    except Exception as e:
        # Non‑critical; log and continue
        print(f"Plotting failed: {e}")

The CSV contains timestamp,price,ma_short,ma_long,signal. The plot is saved as ma_chart.png for quick manual verification; failures in plotting never abort the pipeline.

The Core Logic

Putting the pieces together, the main function orchestrates fetch, transform, compute, label, and persist. Each step is isolated for unit testing and easy replacement.

def main():
    todos = fetch_todos()
    price_df = todos_to_price_series(todos)
    ma_df = add_moving_averages(price_df)
    signal_df = label_signals(ma_df)
    persist_results(signal_df)

if __name__ == "__main__":
    main()

Complete Script

The full runnable script combining all steps:

#!/usr/bin/env python3
"""
From Todos to Trades: a 30‑minute NEPSE moving‑average screener.
Uses JSONPlaceholder todos as a synthetic price feed.
"""

import json
import os
import hashlib
import requests
import pandas as pd
import matplotlib.pyplot as plt
from datetime import datetime, timedelta

CACHE_PATH = "todos_cache.json"

def fetch_todos():
    """Fetch todos from the public API and cache locally."""
    if os.path.exists(CACHE_PATH):
        with open(CACHE_PATH, "r") as f:
            return json.load(f)

    try:
        resp = requests.get(
            "https://jsonplaceholder.typicode.com/todos", timeout=5
        )
        resp.raise_for_status()
        data = resp.json()
        with open(CACHE_PATH, "w") as f:
            json.dump(data, f)
        return data
    except requests.RequestException as e:
        raise RuntimeError(f"Failed to fetch todos: {e}")

def todos_to_price_series(todos):
    """Convert raw todos into a time‑indexed price DataFrame."""
    base = datetime(2023, 1, 1)
    rows = []
    for todo in todos:
        ts = base + timedelta(minutes=todo["id"])
        price = float(todo["userId"])
        rows.append({"timestamp": ts, "price": price})
    df = pd.DataFrame(rows).set_index("timestamp")
    return df.sort_index()

def add_moving_averages(df, short=5, long=20):
    """Add short‑ and long‑window moving averages."""
    df["ma_short"] = df["price"].rolling(window=short, min_periods=1).mean()
    df["ma_long"]  = df["price"].rolling(window=long,  min_periods=1).mean()
    return df

def label_signals(df):
    """Detect golden/death crossovers and label them."""
    df["diff"] = df["ma_short"] - df["ma_long"]
    df["signal"] = 0
    df["prev_diff"] = df["diff"].shift(1).fillna(0)

    buy_mask  = (df["diff"] > 0) & (df["prev_diff"] <= 0)
    sell_mask = (df["diff"] < 0) & (df["prev_diff"] >= 0)

    df.loc[buy_mask,  "signal"] = 1
    df.loc[sell_mask, "signal"] = -1
    return df.drop(columns=["diff", "prev_diff"])

def persist_results(df, csv_path="signals.csv", img_path="ma_chart.png"):
    """Write CSV and optional PNG chart."""
    try:
        df.to_csv(csv_path, index_label="timestamp")
    except OSError as e:
        raise RuntimeError(f"Unable to write CSV: {e}")

    try:
        plt.figure(figsize=(10, 4))
        df["price"].plot(label="Price")
        df["ma_short"].plot(label="MA 5")
        df["ma_long"].plot(label="MA 20")
        plt.legend()
        plt.title("Synthetic Price + Moving Averages")
        plt.savefig(img_path)
        plt.close()
    except Exception as e:
        print(f"Plotting failed: {e}")

def main():
    todos = fetch_todos()
    price_df = todos_to_price_series(todos)
    ma_df = add_moving_averages(price_df)
    signal_df = label_signals(ma_df)
    persist_results(signal_df)

if __name__ == "__main__":
    main()

Expected Output

Running python screener.py creates two files in the working directory:

  • signals.csv – a table with timestamp,price,ma_short,ma_long,signal. Rows where signal is 1 indicate a buy, -1 a sell, and 0 neutral.
  • ma_chart.png – a line chart visualising the synthetic price series and the two moving averages. Buy points appear where the short MA crosses upward.

Limitations and Tradeoffs

This approach is deliberately lightweight, which means:

  • Price data is synthetic; real NEPSE feeds require authentication, rate‑limiting, and timezone handling.
  • Moving averages are computed on a minute‑spaced series derived from id. In production you’d use actual timestamps and possibly adjust for market holidays.
  • Cache invalidation is naive – the script never refreshes the JSON unless the file is manually removed. For nightly runs you’d add a TTL check.
  • We rely on pandas’ in‑memory DataFrames; for >10 M rows you’d switch to Polars or a Spark job.

Frequently Asked Questions

Can I replace the placeholder API with a real NEPSE endpoint?

Yes. Swap fetch_todos for a function that authenticates to the NEPSE REST API, parses the JSON into timestamp and price, and keep the downstream logic unchanged.

Why not use TA‑Lib or a dedicated technical‑analysis library?

TA‑Lib adds a compiled dependency and is overkill for simple moving averages. Pandas’ rolling window is pure Python, easy to test, and sufficient for a lightweight screener.

How do I schedule this script in Airflow?

Wrap the main() call in a BashOperator or PythonOperator. Ensure the DAG sets retries and retry_delay so that network failures (caught as RuntimeError) trigger automatic retries.

What if the CSV write fails due to disk quota?

The persist_results function raises a RuntimeError on OSError. In a containerised job you can catch this exception at the DAG level and alert the ops team.

Can I extend the script to compute more indicators?

Absolutely. Add additional columns after add_moving_averages – e.g., RSI, MACD – and update label_signals to incorporate multi‑indicator logic before persisting.

What I'd Change

If I were to ship this to a production trading desk, I’d replace the ad‑hoc cache with a proper object store (S3 or GCS) and add a TTL‑based invalidation layer so the data stays fresh without manual deletions. I’d also move the moving‑average calculation into a Polars pipeline to shave off ~30 % runtime on multi‑million‑row price histories, and I’d emit the signals via a Kafka topic instead of a static CSV so downstream micro‑services can react in real time. Those upgrades turn a 30‑minute demo into a resilient, scalable component of a live NEPSE trading stack.

Post a Comment

Hi! How can we help you? Send us a message and we'll get back to you.