Task: IEOD Historical Data Pipeline

Convert GDFL Daily CSV files → Script-wise Monthly Parquet files → DuckDB Catalog

Who is this guide for?

This guide is written for developers who may be new to both the Indian stock market and Python. Every concept is explained from scratch. All code is complete and copy-paste ready.

---

Stock Market Glossary

Before writing any code, understand these terms. You will see them everywhere in the codebase.

Term	Plain English Explanation
NSE	National Stock Exchange of India — the marketplace where stocks are bought and sold
Instrument	Anything that can be traded: a stock, an index, a futures contract, or an options contract
OHLCV	Open, High, Low, Close, Volume — the five numbers that describe any time period of price movement
1-minute bar	The OHLCV summary of all trades that happened in one minute. NSE produces ~375 bars per trading day
Market hours	NSE opens at 9:15 AM and closes at 3:30 PM IST. That is 375 minutes per trading day
Spot	The actual stock price right now. Example: RELIANCE share price
Futures (FUT)	A contract to buy/sell a stock at a fixed price on a future date
Options (CE/PE)	The right (but not obligation) to buy (CE) or sell (PE) at a fixed price before expiry
Expiry	The date a futures or options contract ends. NSE has weekly and monthly expiries
Strike	The price written on an options contract
masid	Master ID — a unique number we assign to each instrument in our scr_master database table
scr_master	Our PostgreSQL table that lists every instrument with its masid, name, type, and expiry
GDFL	GDF Ltd — the company that sells us historical 1-minute data in CSV format
IEOD	Intraday End-of-Day — 1-minute OHLCV bar data delivered daily by GDFL
Parquet	A file format that stores data in columns (not rows). Much faster to read for analytics than CSV
DuckDB	An analytics database that can directly query Parquet files using SQL
Bhavcopy	NSE's official daily summary of all end-of-day prices (separate from IEOD)

Think of the Data Flow Like a Newspaper Archive

• GDFL is like the newspaper company — they deliver a new edition every day
• IEOD CSV files are the daily newspapers — each one has all the prices for that day
• PostgreSQL staging is the temporary inbox on your desk
• Parquet files are the archive filing cabinet — organised by symbol and month
• DuckDB catalog is the index card in front of the cabinet — tells you what's where

---

What is GDFL?

GDFL (GDF Ltd) is a market data vendor. They provide historical NSE data as daily CSV files.

What they deliver:

• One CSV file per day per product type
• Contains 1-minute OHLCV bars for ALL instruments traded that day
• File names encode the date: GFDLNFO_CONTRACT_05062024.csv = 5-June-2024 F&O data

Three types of GDFL files:

File Prefix	Contains	Example Filename
GFDLNFO_CONTRACT_	All F&O (Futures + Options)	GFDLNFO_CONTRACT_05062024.csv
GFDLCM_INDICES_	NIFTY, BANKNIFTY, SENSEX etc.	GFDLCM_INDICES_05062024.csv
GFDLCM_STOCK_	All NSE equity stocks	GFDLCM_STOCK_05062024.csv

GDFL CSV column format:

Ticker,Date,Time,Open,High,Low,Close,Volume,OpenInterest
NIFTY24JUN25000CE.NFO,05-06-2024,09:15:00,150.5,155.0,149.0,153.5,12500,450000

GDFL Format Versions

GDFL changed their format three times. The existing code handles all three:

Version	When Used	Key Difference	Prefix Used
v1	Before May 2020	Only NIFTY & BANKNIFTY options	GFDLNFO_NIFTY_BNK_OPT_
v2	May 2020 – Jan 2021	All F&O, monthly expiry written as YYMMM	GFDLNFO_CONTRACT_
v3	Jan 2021 onwards	All F&O, weekly+monthly as DDMMMYY	GFDLNFO_CONTRACT_

Rule of Thumb

• Year 2021 onwards → use version=3
• Year 2020 → use version=2
• Year before 2020 → use version=1

---

What is the Pipeline?

The pipeline has four stages. Think of each stage as a transformation step:

Stage 1 — Ingest: Read the GDFL CSV, figure out which masid each ticker belongs to, and insert the 1-minute bars into PostgreSQL staging tables.

Stage 2 — Pack: Read all bars for a full year from PostgreSQL, write them as monthly Parquet files (one file per symbol per month), then DELETE from PostgreSQL to free space.

Stage 3 — Catalog: Scan all Parquet files and register them in the DuckDB catalog so analytics queries know exactly which files exist and what date ranges they cover.

Why does Stage 1 use PostgreSQL at all? Because the ticker-to-masid lookup is complex SQL. It's easier to do in a proper database than in pure Python. PostgreSQL staging is temporary — it gets cleared after packing.

---

Parquet Store Layout

After Stage 2, files are stored like this:

IEOD_DATA_PATH/               (e.g. D:/ieod/data  or  /home/kss/tradeentry/ieod/data)
├── N/
│   └── NIFTY/
│       ├── 2021/
│       │   ├── 1_JANUARY      ← masid=1, month=January, year=2021
│       │   ├── 1_FEBRUARY
│       │   └── ... (12 files per year)
│       └── 2024/
│           └── 1_JANUARY
├── B/
│   └── BANKNIFTY/
│       └── 2024/
│           └── 2_JANUARY      ← masid=2, month=January
└── R/
    └── RELIANCE/
        └── 2024/
            └── 45_JANUARY     ← masid=45 (example)

Key points:

• First letter of symbol name → folder letter (NIFTY → N/)
• Each file stores ALL contracts of that underlying (spot + futures + options)
• Filter by masid at read time to get individual contract data
• File has NO extension — it is a Parquet file even without .parquet

---

Prerequisites Checklist

Before starting, verify these items are in place:

□ Python venv is active  (backend/venv)
□ duckdb is installed    (pip install duckdb==1.2.2)
□ PostgreSQL is running  (tedb database accessible)
□ scr_master table has all expected symbols populated
□ GDFL CSV files are downloaded and placed in IEOD_SOURCE_PATH
□ IEOD_DATA_PATH directory exists and has write permission
□ backend/.env has correct LOCAL_IEOD_DATA_PATH and LOCAL_IEOD_SOURCE_PATH set

Check your paths:

# Run this in a Jupyter notebook (backend/scratch/)
import sys
from pathlib import Path

BACKEND_DIR = Path("..").resolve()
sys.path.insert(0, str(BACKEND_DIR))

from app.config import Config

print("IEOD Source Path :", Config.IEOD_SOURCE_PATH)
print("IEOD Data Path   :", Config.IEOD_DATA_PATH)
print("Source exists    :", Path(Config.IEOD_SOURCE_PATH).exists())
print("Data exists      :", Path(Config.IEOD_DATA_PATH).exists())

---

Task 1 — Understand the Existing Code

The nseIEOD.py module in backend/nseieod/ already has the complete pipeline. Read this section to understand what each class and method does before writing any code.

The IEod Class

# backend/nseieod/nseIEOD.py

from nseieod.nseIEOD import IEod
import datetime

mu = IEod(
    fromdate = datetime.date(2024, 1, 1),   # Start date for this run
    todate   = datetime.date(2024, 3, 31),   # End date for this run
    path     = "/path/to/ieod/source",       # GDFL CSV source folder
    version  = 3,                            # GDFL format version (see table above)
    fo_prefix= "GFDLNFO_CONTRACT_",          # File name prefix for F&O files
    datapath = "/path/to/ieod/data",         # Parquet output folder
)

Methods you will use:

Method	What it does
mu.update_folder()	Reads all GDFL CSV files in fromdate..todate range → inserts into PostgreSQL staging
mu.pack(year)	Moves all data for that year from PostgreSQL → Parquet files, then deletes from PostgreSQL

The Con_Price_Getter Class

This class is for reading prices back after the data is packed. You do not need to modify it.

from nseieod.nseIEOD import Con_Price_Getter

cp = Con_Price_Getter(
    script  = "NIFTY",                   # Symbol name
    datapath= "/path/to/ieod/data",      # Parquet data folder
    fromdate= "2024-01-01",
    todate  = "2024-03-31",
    opttype = "FUT",                     # "SPOT", "FUT", "CE", or "PE"
    expiry  = "M1",                      # M1 = nearest monthly expiry
)
df = cp.getprices()

---

Task 2 — Ingest GDFL CSVs into PostgreSQL Staging

This task reads each daily CSV file and writes 1-minute bars into the PostgreSQL staging tables scr_nsefo (F&O) and scr_nseeq (equity/index).

Step 2.1 — Create the Ingest Script

Create a new file: backend/nseieod/run_ingest.py

"""
run_ingest.py
─────────────
Reads GDFL F&O CSV files from IEOD_SOURCE_PATH for a given date range
and inserts 1-minute bars into PostgreSQL staging (scr_nsefo / scr_nseeq).

Run once per day (or in bulk for historical backfill).

Usage:
    python run_ingest.py --from 2024-01-01 --to 2024-01-31
    python run_ingest.py --from 2024-01-01 --to 2024-01-31 --version 3
"""

import sys
import argparse
import datetime
from pathlib import Path

# ── Make sure backend/ is on sys.path ─────────────────────────────────────
_backend = str(Path(__file__).resolve().parents[1])
if _backend not in sys.path:
    sys.path.insert(0, _backend)

from app.config import Config
from nseieod.nseIEOD import IEod


def ingest(from_date: str, to_date: str, version: int):
    """
    Ingest GDFL CSVs into PostgreSQL staging tables.

    Parameters
    ----------
    from_date : str   'YYYY-MM-DD'
    to_date   : str   'YYYY-MM-DD'
    version   : int   1, 2, or 3 (see GDFL Format Versions table)
    """
    mfrom = datetime.date.fromisoformat(from_date)
    mto   = datetime.date.fromisoformat(to_date)

    # Choose prefix based on version
    if version == 1:
        prefix = "GFDLNFO_NIFTY_BNK_OPT_"
    else:
        prefix = "GFDLNFO_CONTRACT_"

    print(f"[ingest] {from_date} → {to_date}  version={version}  prefix={prefix}")
    print(f"[ingest] Source : {Config.IEOD_SOURCE_PATH}")
    print(f"[ingest] Data   : {Config.IEOD_DATA_PATH}")

    mu = IEod(
        fromdate  = mfrom,
        todate    = mto,
        path      = Config.IEOD_SOURCE_PATH,
        version   = version,
        fo_prefix = prefix,
        datapath  = Config.IEOD_DATA_PATH,
    )

    mu.update_folder()
    print("[ingest] Done. Run run_pack.py to convert to Parquet.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Ingest GDFL CSVs into PostgreSQL")
    parser.add_argument("--from", dest="from_date", required=True,
                        help="Start date YYYY-MM-DD")
    parser.add_argument("--to",   dest="to_date",   required=True,
                        help="End date YYYY-MM-DD")
    parser.add_argument("--version", type=int, default=3,
                        help="GDFL version: 1, 2, or 3 (default: 3)")
    args = parser.parse_args()

    ingest(args.from_date, args.to_date, args.version)

Step 2.2 — Run the Ingest

# Activate venv first
cd backend
source venv/bin/activate          # Linux/Mac
# or: venv\Scripts\activate       # Windows

# Ingest one month (2024, version 3)
python nseieod/run_ingest.py --from 2024-01-01 --to 2024-01-31 --version 3

# Ingest historical 2019 data (version 2)
python nseieod/run_ingest.py --from 2019-01-01 --to 2019-12-31 --version 2

# Ingest very old 2015 data (version 1)
python nseieod/run_ingest.py --from 2015-01-01 --to 2015-12-31 --version 1

Step 2.3 — Verify the Staging Data

Run this in a notebook after ingest to confirm data landed in PostgreSQL:

from nseeod import pgs
import pandas as pd

# How many rows in staging for a specific date?
check_date = "2024-01-15"

fo_count = pgs.getsqlValue(
    f"SELECT COUNT(*) FROM scr_nsefo WHERE DATE(sdate) = '{check_date}'"
)
print(f"F&O bars for {check_date}: {fo_count}")

# Show a sample
df = pgs.read_sql(
    f"SELECT masid, sdate, sopen, shigh, slow, sclose, svolume "
    f"FROM scr_nsefo WHERE DATE(sdate) = '{check_date}' LIMIT 10"
)
display(df)

Expected results:

• A busy trading day should have 375 rows per instrument (one per minute 09:15–15:29)
• masid values should all be non-null and exist in scr_master
• If count is 0, the CSV file may be missing or named differently

---

Task 3 — Pack PostgreSQL Staging → Parquet Files

The pack step:

1. Reads all 1-minute bars for a full year from scr_nsefo and scr_nseeq
2. Groups them by symbol and month
3. Writes each group as a Parquet file
4. Deletes the rows from PostgreSQL

Important

pack(year) permanently DELETES the year's data from PostgreSQL staging after writing Parquet. Always verify the Parquet files exist before assuming the data is safe.

Step 3.1 — Create the Pack Script

Create a new file: backend/nseieod/run_pack.py

"""
run_pack.py
───────────
Moves 1-minute bar data for a full year from PostgreSQL staging tables
(scr_nsefo, scr_nseeq) into month-partitioned Parquet files, then
clears the staging tables for that year.

Run AFTER run_ingest.py has completed for the full year.
Never run for a partial year — pack one complete year at a time.

Usage:
    python run_pack.py --year 2024
    python run_pack.py --year 2024 --dry-run   (shows what would be packed, no writes)
"""

import sys
import argparse
from pathlib import Path

_backend = str(Path(__file__).resolve().parents[1])
if _backend not in sys.path:
    sys.path.insert(0, _backend)

from app.config import Config
from nseieod.nseIEOD import IEod
from nseeod import pgs
import datetime


def preview_pack(year: int):
    """Show how many rows would be packed without actually writing."""
    mfrom = f"{year}-01-01"
    mto   = f"{year}-12-31 23:59:59"

    fo_count = pgs.getsqlValue(
        f"SELECT COUNT(*) FROM scr_nsefo WHERE sdate BETWEEN '{mfrom}' AND '{mto}'"
    )
    eq_count = pgs.getsqlValue(
        f"SELECT COUNT(*) FROM scr_nseeq WHERE sdate BETWEEN '{mfrom}' AND '{mto}'"
    )
    print(f"[dry-run] Year {year}:")
    print(f"  scr_nsefo rows to pack : {fo_count:,}")
    print(f"  scr_nseeq rows to pack : {eq_count:,}")
    print(f"  Total                  : {(fo_count or 0) + (eq_count or 0):,}")

    if (fo_count or 0) + (eq_count or 0) == 0:
        print("  ⚠  No data found! Did ingest run for this year?")
    else:
        print("  ✓  Data found. Run without --dry-run to pack.")


def run_pack(year: int):
    """Pack a full year from PostgreSQL → Parquet."""
    print(f"[pack] Starting pack for year {year}")
    print(f"[pack] Output : {Config.IEOD_DATA_PATH}")

    # IEod needs fromdate/todate but pack() ignores them — year is what matters
    mu = IEod(
        fromdate  = datetime.date(year, 1, 1),
        todate    = datetime.date(year, 12, 31),
        path      = Config.IEOD_SOURCE_PATH,
        version   = 3,                             # version only needed for update_folder
        fo_prefix = "GFDLNFO_CONTRACT_",
        datapath  = Config.IEOD_DATA_PATH,
    )

    mu.pack(year)
    print(f"[pack] Done. Parquet files written and PostgreSQL staging cleared for {year}.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Pack PostgreSQL staging → Parquet")
    parser.add_argument("--year",    type=int, required=True, help="Year to pack, e.g. 2024")
    parser.add_argument("--dry-run", action="store_true",
                        help="Preview what would be packed without writing")
    args = parser.parse_args()

    if args.dry_run:
        preview_pack(args.year)
    else:
        run_pack(args.year)

Step 3.2 — Run the Pack

# Always preview first
python nseieod/run_pack.py --year 2024 --dry-run

# If preview looks correct, run the actual pack
python nseieod/run_pack.py --year 2024

Step 3.3 — Verify Parquet Files Were Created

# In a notebook
from pathlib import Path
from app.config import Config
import pandas as pd

data_root = Path(Config.IEOD_DATA_PATH)

# Count all Parquet files created for 2024
parquet_files = list(data_root.glob("*/**/2024/*"))
print(f"Parquet files created for 2024: {len(parquet_files)}")

# Sample: read NIFTY January 2024
nifty_jan = data_root / "N" / "NIFTY" / "2024" / "1_JANUARY"
if nifty_jan.exists():
    df = pd.read_parquet(nifty_jan, engine='pyarrow')
    print(f"\nNIFTY January 2024:")
    print(f"  Total rows  : {len(df):,}")
    print(f"  masid values: {sorted(df['masid'].unique())[:10]}")
    print(f"  Date range  : {df['sdate'].min()} → {df['sdate'].max()}")
else:
    print("⚠ NIFTY January 2024 Parquet file not found!")

Expected output:

Parquet files created for 2024: 240+  (depends on how many symbols have data)

NIFTY January 2024:
  Total rows  : 450,000+
  masid values: [1, 3, 4, 5, 6, ...]
  Date range  : 2024-01-01 09:15:00+05:30 → 2024-01-31 15:29:00+05:30

---

Task 4 — Build the DuckDB Catalog

The DuckDB catalog is an index of all Parquet files. It stores: which files exist, what date range they cover, how many rows they have, and any quality issues.

Why Do We Need a Catalog?

Without a catalog, every query must scan the entire directory tree to find relevant files. With a catalog, DuckDB can jump straight to the right files in milliseconds.

Step 4.1 — Create the DuckDB Package

Create these files (the complete DuckDB sub-package):

File: backend/nseieod/duckdb/__init__.py

from .catalog_db    import get_catalog_con
from .catalog_scan  import scan_parquet_tree

__all__ = ["get_catalog_con", "scan_parquet_tree"]

---

File: backend/nseieod/duckdb/catalog_db.py

"""
catalog_db.py — DuckDB catalog schema + connection factory.

The catalog is a single .db file that indexes every Parquet file in the
Parquet store. It stores file metadata (path, masid, date range, row count)
but NEVER duplicates scr_master data (symbol names, expiry, strikes).

The catalog lives at:
    {IEOD_DATA_PATH}/_catalog/catalog.db
"""

import duckdb
from pathlib import Path
from app.config import Config

CATALOG_PATH = Path(Config.IEOD_DATA_PATH) / "_catalog" / "catalog.db"


def get_catalog_con(read_only: bool = False) -> duckdb.DuckDBPyConnection:
    """
    Open (or create) the DuckDB catalog.

    Parameters
    ----------
    read_only : bool
        Set True for query-only access (analytics, notebooks).
        Set False (default) when scanning / updating the catalog.

    Returns
    -------
    duckdb.DuckDBPyConnection
        Always call .close() when done.
    """
    CATALOG_PATH.parent.mkdir(parents=True, exist_ok=True)
    con = duckdb.connect(str(CATALOG_PATH), read_only=read_only)
    _ensure_schema(con)
    return con


def _ensure_schema(con: duckdb.DuckDBPyConnection) -> None:
    """Create catalog tables if they do not exist yet."""

    # One row per Parquet file
    con.execute("""
        CREATE TABLE IF NOT EXISTS files (
            file_id       INTEGER PRIMARY KEY,
            masid         INTEGER  NOT NULL,   -- links to scr_master.masid in PostgreSQL
            root_masid    INTEGER  NOT NULL,   -- underlying symbol masid (e.g. NIFTY = 1)
            tier          VARCHAR  NOT NULL,   -- 'active' (2015+) or 'archive' (pre-2015)
            year          INTEGER  NOT NULL,
            month_name    VARCHAR  NOT NULL,   -- 'JANUARY' to 'DECEMBER'
            file_path     VARCHAR  NOT NULL UNIQUE,
            row_count     BIGINT,
            first_ts      TIMESTAMPTZ,         -- earliest bar timestamp in file
            last_ts       TIMESTAMPTZ,         -- latest bar timestamp in file
            file_size_kb  INTEGER,
            scanned_at    TIMESTAMPTZ DEFAULT now(),
            quality       VARCHAR  DEFAULT 'unknown'  -- 'ok' | 'gaps' | 'sparse' | 'empty'
        )
    """)

    # Known date ranges where data is missing or incomplete
    con.execute("""
        CREATE TABLE IF NOT EXISTS gaps (
            gap_id       INTEGER PRIMARY KEY,
            masid        INTEGER NOT NULL,
            gap_start    TIMESTAMPTZ NOT NULL,
            gap_end      TIMESTAMPTZ NOT NULL,
            gap_minutes  INTEGER,
            gap_type     VARCHAR,   -- 'intraday' | 'multiday'
            file_id      INTEGER REFERENCES files(file_id)
        )
    """)

    # Sequences for auto-incrementing IDs
    con.execute("CREATE SEQUENCE IF NOT EXISTS seq_file_id START 1")
    con.execute("CREATE SEQUENCE IF NOT EXISTS seq_gap_id  START 1")

---

File: backend/nseieod/duckdb/catalog_scan.py

"""
catalog_scan.py — Scans the Parquet tree and populates catalog.files.

Walk the directory tree under IEOD_DATA_PATH:
    {letter}/{symbol}/{year}/{root_masid}_{MONTH_NAME}

For each file found:
- Read basic stats (row count, first/last timestamp)
- Insert a new row in catalog.files  (or update if file changed size)

Run after every pack() operation.
"""

import os
import duckdb
from pathlib import Path
from app.config import Config
from .catalog_db import get_catalog_con


# All valid month names in uppercase
VALID_MONTHS = {
    "JANUARY", "FEBRUARY", "MARCH", "APRIL", "MAY", "JUNE",
    "JULY", "AUGUST", "SEPTEMBER", "OCTOBER", "NOVEMBER", "DECEMBER"
}


def scan_parquet_tree(
    tier:       str  = "active",
    root_path:  str  = None,
    verbose:    bool = True
) -> dict:
    """
    Scan all Parquet files under root_path and update catalog.files.

    Parameters
    ----------
    tier       : 'active' (default) or 'archive'
    root_path  : Directory to scan. Defaults to Config.IEOD_DATA_PATH.
    verbose    : Print progress as files are scanned.

    Returns
    -------
    dict with keys: scanned, skipped, errors
    """
    root    = Path(root_path or Config.IEOD_DATA_PATH)
    con     = get_catalog_con()

    counts  = {"scanned": 0, "skipped": 0, "errors": 0}

    # Walk: root / {A-Z} / {symbol} / {year} / {masid}_{MONTH}
    for letter_dir in sorted(root.iterdir()):
        if not letter_dir.is_dir():
            continue
        if letter_dir.name.startswith("_"):
            # Skip _catalog/ and similar internal folders
            continue

        for symbol_dir in sorted(letter_dir.iterdir()):
            if not symbol_dir.is_dir():
                continue

            for year_dir in sorted(symbol_dir.iterdir()):
                if not year_dir.is_dir() or not year_dir.name.isdigit():
                    continue

                year = int(year_dir.name)

                for pfile in sorted(year_dir.iterdir()):
                    # Skip non-Parquet files (.csv, .log, etc.)
                    if pfile.suffix.lower() in (".csv", ".log", ".txt"):
                        continue

                    # Parse filename: {root_masid}_{MONTH_NAME}
                    parts = pfile.name.split("_", 1)
                    if len(parts) != 2:
                        continue

                    try:
                        root_masid = int(parts[0])
                    except ValueError:
                        continue

                    month_name = parts[1].upper()
                    if month_name not in VALID_MONTHS:
                        continue

                    file_path    = pfile.as_posix()
                    current_size = pfile.stat().st_size // 1024

                    # Check if already catalogued with same size (no change)
                    existing = con.execute(
                        "SELECT file_id, file_size_kb FROM files WHERE file_path = ?",
                        [file_path]
                    ).fetchone()

                    if existing and existing[1] == current_size:
                        counts["skipped"] += 1
                        continue

                    # Read stats from the Parquet file
                    try:
                        tmp = duckdb.connect()
                        stats = tmp.execute(f"""
                            SELECT
                                masid,
                                COUNT(*)   AS row_count,
                                MIN(sdate) AS first_ts,
                                MAX(sdate) AS last_ts
                            FROM read_parquet('{file_path}')
                            GROUP BY masid
                        """).fetchall()
                        tmp.close()
                    except Exception as e:
                        if verbose:
                            print(f"  [ERROR] {pfile.name}: {e}")
                        counts["errors"] += 1
                        continue

                    # Upsert one row per masid in this file
                    for masid, row_count, first_ts, last_ts in stats:
                        if existing:
                            con.execute("""
                                UPDATE files SET
                                    row_count    = ?,
                                    first_ts     = ?,
                                    last_ts      = ?,
                                    file_size_kb = ?,
                                    scanned_at   = now(),
                                    quality      = 'unknown'
                                WHERE file_path = ? AND masid = ?
                            """, [row_count, first_ts, last_ts,
                                  current_size, file_path, masid])
                        else:
                            con.execute("""
                                INSERT INTO files (
                                    file_id, masid, root_masid, tier, year,
                                    month_name, file_path, row_count,
                                    first_ts, last_ts, file_size_kb
                                ) VALUES (
                                    nextval('seq_file_id'), ?, ?, ?, ?,
                                    ?, ?, ?, ?, ?, ?
                                )
                            """, [masid, root_masid, tier, year, month_name,
                                  file_path, row_count, first_ts, last_ts,
                                  current_size])

                    counts["scanned"] += 1
                    if verbose and counts["scanned"] % 50 == 0:
                        print(f"  [{tier}] Scanned {counts['scanned']} files ...")

    con.close()

    print(f"\n[scan_parquet_tree] Finished tier='{tier}'")
    print(f"  Scanned : {counts['scanned']}")
    print(f"  Skipped : {counts['skipped']}  (already up-to-date)")
    print(f"  Errors  : {counts['errors']}")

    return counts

Step 4.2 — Run the Catalog Scan

Create backend/nseieod/run_catalog_scan.py:

"""
run_catalog_scan.py
───────────────────
Scans all Parquet files and updates the DuckDB catalog.
Run this after every pack() operation.

Usage:
    python nseieod/run_catalog_scan.py
    python nseieod/run_catalog_scan.py --tier archive --path /archive/ieod
"""

import sys
import argparse
from pathlib import Path

_backend = str(Path(__file__).resolve().parents[1])
if _backend not in sys.path:
    sys.path.insert(0, _backend)

from nseieod.duckdb.catalog_scan import scan_parquet_tree


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Scan Parquet tree → update DuckDB catalog")
    parser.add_argument("--tier", default="active",
                        choices=["active", "archive"],
                        help="Which tier to scan (default: active)")
    parser.add_argument("--path", default=None,
                        help="Override IEOD_DATA_PATH (for archive scan)")
    args = parser.parse_args()

    scan_parquet_tree(tier=args.tier, root_path=args.path, verbose=True)

# Run catalog scan
python nseieod/run_catalog_scan.py

# For archive tier
python nseieod/run_catalog_scan.py --tier archive --path /archive/ieod

Step 4.3 — Inspect the Catalog

Run this in a notebook to see what the catalog contains:

from nseieod.duckdb.catalog_db import get_catalog_con
import pandas as pd

con = get_catalog_con(read_only=True)

# Summary by tier and year
summary = con.execute("""
    SELECT
        tier,
        year,
        COUNT(DISTINCT masid)  AS instruments,
        COUNT(*)               AS parquet_files,
        SUM(row_count) / 1e6   AS total_rows_millions,
        MIN(first_ts)::DATE    AS data_from,
        MAX(last_ts)::DATE     AS data_to,
        SUM(file_size_kb)/1024 AS total_size_MB
    FROM files
    GROUP BY tier, year
    ORDER BY year DESC
""").df()

con.close()
display(summary)

---

Task 5 — Historical Backfill (Bulk Run)

If you need to process years of historical data (e.g. 2008–2023), run the pipeline year by year. Do NOT try to do all years in one step — pack one year at a time.

Step 5.1 — Backfill Script

Create backend/nseieod/run_backfill.py:

"""
run_backfill.py
───────────────
Runs the full pipeline (ingest → pack → catalog) for a range of years.
Use this for historical backfill only.

IMPORTANT:
  - Run from the backend/ directory with venv active
  - Each year will: ingest CSVs → pack to Parquet → update catalog
  - Expect this to take several hours for 10+ years of data
  - Monitor disk space: each year of F&O data ≈ 1–3 GB as Parquet

Usage:
    python nseieod/run_backfill.py --from-year 2015 --to-year 2020
"""

import sys
import argparse
import datetime
from pathlib import Path

_backend = str(Path(__file__).resolve().parents[1])
if _backend not in sys.path:
    sys.path.insert(0, _backend)

from app.config import Config
from nseieod.nseIEOD import IEod
from nseieod.duckdb.catalog_scan import scan_parquet_tree


def get_version(year: int) -> tuple[int, str]:
    """Return (version, prefix) for a given year."""
    if year < 2020:
        return 1, "GFDLNFO_NIFTY_BNK_OPT_"
    elif year == 2020:
        return 2, "GFDLNFO_CONTRACT_"
    else:
        return 3, "GFDLNFO_CONTRACT_"


def backfill(from_year: int, to_year: int, skip_ingest: bool = False):
    for year in range(from_year, to_year + 1):
        print(f"\n{'='*60}")
        print(f"  Processing year {year}")
        print(f"{'='*60}")

        version, prefix = get_version(year)

        mu = IEod(
            fromdate  = datetime.date(year, 1, 1),
            todate    = datetime.date(year, 12, 31),
            path      = Config.IEOD_SOURCE_PATH,
            version   = version,
            fo_prefix = prefix,
            datapath  = Config.IEOD_DATA_PATH,
        )

        if not skip_ingest:
            print(f"\n[{year}] Stage 1: Ingesting CSVs → PostgreSQL ...")
            mu.update_folder()

        print(f"\n[{year}] Stage 2: Packing PostgreSQL → Parquet ...")
        mu.pack(year)

        print(f"\n[{year}] Stage 3: Updating DuckDB catalog ...")
        scan_parquet_tree(tier="active", verbose=False)

        print(f"\n[{year}] ✓ Complete")


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="Backfill historical IEOD data")
    parser.add_argument("--from-year", type=int, required=True)
    parser.add_argument("--to-year",   type=int, required=True)
    parser.add_argument("--skip-ingest", action="store_true",
                        help="Skip CSV ingest (useful if CSVs were already ingested)")
    args = parser.parse_args()

    backfill(args.from_year, args.to_year, args.skip_ingest)

# Example: backfill 2015–2020
python nseieod/run_backfill.py --from-year 2015 --to-year 2020

---

Task 6 — Daily Update (Ongoing)

After the historical backfill is done, run this every trading day to keep data current.

Create backend/nseieod/run_daily.py:

"""
run_daily.py
────────────
Runs the daily IEOD update pipeline:
  1. Ingest today's GDFL CSV into PostgreSQL staging
  2. Every month-end (or year-end): pack to Parquet + update catalog

Schedule this via cron on the VPS.

Usage:
    python nseieod/run_daily.py
    python nseieod/run_daily.py --date 2024-06-05   (override date)
"""

import sys
import datetime
import argparse
from pathlib import Path

_backend = str(Path(__file__).resolve().parents[1])
if _backend not in sys.path:
    sys.path.insert(0, _backend)

from app.config import Config
from nseieod.nseIEOD import IEod
from nseieod.duckdb.catalog_scan import scan_parquet_tree


def daily_update(target_date: datetime.date):
    year = target_date.year
    # version 3 for 2021+, version 2 for 2020, version 1 for before
    if year >= 2021:
        version, prefix = 3, "GFDLNFO_CONTRACT_"
    elif year == 2020:
        version, prefix = 2, "GFDLNFO_CONTRACT_"
    else:
        version, prefix = 1, "GFDLNFO_NIFTY_BNK_OPT_"

    print(f"[daily] Processing {target_date}")

    mu = IEod(
        fromdate  = target_date,
        todate    = target_date,
        path      = Config.IEOD_SOURCE_PATH,
        version   = version,
        fo_prefix = prefix,
        datapath  = Config.IEOD_DATA_PATH,
    )

    # Ingest today's CSV
    mu.update_folder()

    # If it is month-end OR year-end, pack and update catalog
    next_day = target_date + datetime.timedelta(days=1)
    is_month_end = (next_day.month != target_date.month)
    is_year_end  = (target_date.month == 12 and is_month_end)

    if is_year_end:
        print(f"[daily] Year-end detected. Packing {year} ...")
        mu.pack(year)
        scan_parquet_tree(tier="active", verbose=False)
        print(f"[daily] ✓ Year {year} packed and catalogued.")
    else:
        print(f"[daily] Ingest complete. Pack will run at year-end.")


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--date", default=None,
                        help="Override date YYYY-MM-DD (default: today)")
    args = parser.parse_args()

    target = (datetime.date.fromisoformat(args.date)
              if args.date else datetime.date.today())

    daily_update(target)

Add to VPS cron (runs at 4:30 PM every weekday):

# crontab -e
30 16 * * 1-5 /home/kss/tradeentry/backend/venv/bin/python /home/kss/tradeentry/backend/nseieod/run_daily.py >> /home/kss/tradeentry/logs/ieod_daily.log 2>&1

---

Task 7 — Verify Data Quality

After ingesting and packing, run these checks to confirm data is correct.

Create backend/scratch/duckdb_quality_check.ipynb:

# Cell 1: Setup
import sys
from pathlib import Path
BACKEND_DIR = Path("..").resolve()
sys.path.insert(0, str(BACKEND_DIR))

from nseieod.duckdb.catalog_db import get_catalog_con
import pandas as pd

# Cell 2: Catalog summary
con = get_catalog_con(read_only=True)

summary = con.execute("""
    SELECT
        tier,
        year,
        COUNT(DISTINCT masid)       AS instruments,
        COUNT(*)                    AS parquet_files,
        ROUND(SUM(row_count) / 1e6, 1) AS total_rows_M,
        MIN(first_ts)::DATE         AS data_from,
        MAX(last_ts)::DATE          AS data_to,
        ROUND(SUM(file_size_kb)/1024.0, 0) AS size_MB
    FROM files
    GROUP BY tier, year
    ORDER BY year DESC
""").df()

display(summary)

# Cell 3: Find sparse files (likely incomplete data)
sparse = con.execute("""
    SELECT file_path, masid, year, month_name, row_count
    FROM files
    WHERE row_count < 1000    -- healthy month should have 5000+ rows
    ORDER BY row_count ASC
    LIMIT 20
""").df()

print(f"Sparse files (< 1000 rows): {len(sparse)}")
display(sparse)

# Cell 4: Check a specific symbol's date coverage
# masid=1 is NIFTY spot
coverage = con.execute("""
    SELECT year, month_name, row_count, first_ts::DATE, last_ts::DATE, quality
    FROM files
    WHERE masid = 1
    ORDER BY year, month_name
""").df()

display(coverage)
con.close()

---

Troubleshooting

"File not Found" during ingest

2024-01-05 File not Found!!

Cause: The CSV file for that date is missing from IEOD_SOURCE_PATH.
Fix: Download the missing GDFL file and place it in the source folder.

---

"Script Not Found" during ingest

Script id Should not Zero!!!

Cause: The Ticker from the CSV does not match any mname in scr_master.
Fix: Check scripts.csv (auto-created during ingest) to see which scripts are missing. Insert them into scr_master manually or ask the project lead.

---

Parquet file count is lower than expected after pack

Cause: Some symbols had zero data for that year (e.g. newly listed stocks).
Fix: This is normal. Use the catalog to see which symbols have data for which years.

---

DuckDB error: "Catalog format not compatible"

duckdb.IOException: Catalog file format version mismatch

Cause: DuckDB was upgraded and the .db file format changed.
Fix:

# Delete old catalog and rebuild
rm /ieod/_catalog/catalog.db
python nseieod/run_catalog_scan.py

---

Ingest runs but PostgreSQL staging is still empty

Cause: update_folder() shows no error but the source folder path is wrong.
Fix: Add a debug print and verify the path:

from pathlib import Path
from app.config import Config

source = Path(Config.IEOD_SOURCE_PATH)
print("Source path:", source)
print("Exists:", source.exists())
print("CSV files found:", list(source.glob("GFDLNFO_CONTRACT_*.csv"))[:5])

---

Completion Checklist

Use this checklist to confirm the pipeline is fully working before marking the task done.

Setup
□ duckdb==1.2.2 added to requirements.txt and installed
□ backend/app/config.py has IEOD_DATA_PATH and IEOD_SOURCE_PATH
□ backend/.env has correct LOCAL_IEOD_DATA_PATH and LOCAL_IEOD_SOURCE_PATH

Ingest
□ run_ingest.py created and tested for at least one date
□ scr_nsefo and scr_nseeq have rows after ingest
□ No "Script Not Found" errors for major symbols (NIFTY, BANKNIFTY, RELIANCE)

Pack
□ run_pack.py created and tested for at least one year
□ Parquet files exist under IEOD_DATA_PATH/{letter}/{symbol}/{year}/
□ NIFTY January file has 300,000+ rows (spot + futures + options combined)
□ PostgreSQL staging tables are empty after pack (data moved to Parquet)

DuckDB Catalog
□ nseieod/duckdb/__init__.py created
□ nseieod/duckdb/catalog_db.py created
□ nseieod/duckdb/catalog_scan.py created
□ run_catalog_scan.py created and tested
□ catalog.db exists at {IEOD_DATA_PATH}/_catalog/catalog.db
□ Catalog summary query returns correct row counts

Data Quality
□ Quality check notebook runs without errors
□ No critical symbols have zero rows
□ Sparse file count is reasonable (minor gaps expected for holidays)

Automation
□ run_daily.py created and tested for one date
□ Cron job scheduled on VPS (or Windows Task Scheduler on local)
□ Daily log file is being written

Historical Backfill
□ At least 2 years of historical data is packed and in catalog
□ ieod_query.ipynb confirms data is readable via get_ieod_data()
□ DuckDB direct query returns same results as get_ieod_data()

---

Files Created in This Task

backend/
  nseieod/
    duckdb/
      __init__.py            ← Package exports
      catalog_db.py          ← Schema + connection factory
      catalog_scan.py        ← Scans Parquet tree → catalog.files
    run_ingest.py            ← Stage 1: CSV → PostgreSQL
    run_pack.py              ← Stage 2: PostgreSQL → Parquet
    run_catalog_scan.py      ← Stage 3: Parquet → DuckDB catalog
    run_daily.py             ← Daily automation script
    run_backfill.py          ← Historical bulk backfill
  scratch/
    duckdb_quality_check.ipynb   ← Data quality verification notebook

All scripts in nseieod/ can be run directly from the command line. All configuration (paths, DB credentials) comes from app/config.py which reads backend/.env. No hardcoded paths anywhere.