Build a Fail-Closed NSE Data Pipeline
Build a safe CLI that validates, caches, and replays NSE market candles.
Introduction
30 Second Summary
Market prices can look trustworthy even when updates stop or values arrive broken. One bad number can turn an automated decision into a costly one.
In this project, you will build a command-line service that turns recent NSE market data into a validated candle history. A fail-closed safety gate prevents the research provider from authorizing future live orders.
What You'll Build
The finished demo shows an ordered SBIN candle history before the live-data check stops an unsafe provider.
By the end of this project, you'll have:
- A validated SBIN feed that prints recent 5-minute candles in one normalized shape.
- A duplicate-free candle history that survives repeated downloads in SQLite. You can replay it in timestamp order whenever you need deterministic input.
- A live-data safety check that visibly blocks yfinance from live decisions. The project also preserves your confirmed ₹50 daily-loss cap in config.toml for later risk controls.
- Secret Mission: Simulate a nominally live feed becoming stale. Watch the five-second safety gate reject it.
Are there any prerequisites?
None. This guide verifies every required tool from scratch on Linux.
You need an internet connection. You do not need a broker account.
Before We Start
This first checkpoint locks in the purpose of your fail-closed NSE data pipeline before any setup begins. Every future strategy will receive the same validated candles while unsafe research data stays blocked from live trading decisions.
Set Up a Reproducible Python Workspace
Every later candle check depends on a Python workspace you can recreate. Version drift can make the same pipeline behave differently between machines.
A pinned interpreter keeps language behavior stable. A virtual environment holds the project’s exact dependencies away from global packages.
In this step, get ready to:
- Prepare the project workspace in Visual Studio Code.
- Install uv 0.13.6 with a managed Python 3.11 environment.
- Create a documented dependency baseline for the pipeline.
Prepare the Visual Studio Code workspace
Visual Studio Code keeps your project files in one workspace. Its integrated terminal runs commands from that same folder.
- Press the Super key to open your Linux application search.
- Search for your terminal application.
- Select the terminal application from the results.
- Check whether Visual Studio Code is available by running this command:
code --versionWhat Does This Command Check?
The command asks the Visual Studio Code command-line interface for its installed version. A version response confirms that the editor is available from your terminal.
✔️ I see version information
Good start. Visual Studio Code is installed with its terminal command available.
ⓧ Command not found
The primary Linux installation path uses Snap. Your distribution may provide a different official package path.
- Install Visual Studio Code through Snap by running this command:
sudo snap install --classic codeWhat Does This Command Do?
Snap installs the Visual Studio Code package with classic confinement. The elevated permission lets the package manager complete the system installation.
The terminal should confirm that Visual Studio Code installed successfully.
- Use the official Visual Studio Code Linux setup page if your distribution does not support Snap.
- Rerun the version check shown above after the installation completes.
Installation Did Not Complete?
Confirm that your account can run commands with elevated permission. Use the distribution-specific package instructions when Snap is unavailable.
Ask for targeted help with installing Visual Studio Code on your Linux distribution.
- Press the Super key to open your Linux application search.
- Type Visual Studio Code into the search field.
- Select Visual Studio Code from the results.
- Select File in the top menu.
- Select Open Folder.
- Select your Desktop location.
- Select New Folder.
- Enter fail-closed-nse-data-pipeline as the folder name.
The folder picker now shows your empty project folder on the Desktop.
- Select the fail-closed-nse-data-pipeline folder.
- Select Select Folder.
- Select Yes, I trust the authors if the Workspace Trust dialog appears.
Your Explorer now shows fail-closed-nse-data-pipeline. The editor and terminal can now work from the same project location.
Create the managed Python environment
uv manages the pinned Python release plus the project environment. This gives the workspace a reproducible interpreter without depending on the Linux system Python.
- Select View in the Visual Studio Code top menu.
- Select Terminal.
The integrated terminal opens inside fail-closed-nse-data-pipeline.
- Check the installed uv release by running this command:
uv --versionWhat Does This Command Check?
The command prints the uv release available to the current shell. The project uses the pinned 0.13.6 release.
✔️ I see version 0.13.6
uv 0.13.6 is ready to manage the project interpreter.
ⓧ I see another version
Install the pinned release so this workspace uses the tested uv behavior.
- Install uv 0.13.6 by running this command:
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | shWhat Does This Command Do?
The command downloads the official uv 0.13.6 installer. It passes the installer script to your shell.
- Open a fresh integrated terminal with the terminal panel’s plus button.
- Rerun the uv version check shown above.
The Pinned Release Is Missing?
Confirm that the installer completed before opening the fresh terminal. Check your internet connection if the download stopped early.
Ask for help with installing the pinned uv release.
ⓧ Command not found
Install the pinned release before creating the Python environment.
- Install uv 0.13.6 by running this command:
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | shWhat Does This Command Do?
The command downloads the official uv 0.13.6 installer. It installs the tool for your Linux user.
- Open a fresh integrated terminal with the terminal panel’s plus button.
- Rerun the uv version check shown above.
uv Is Still Unavailable?
Confirm that the installation finished without a network failure. A fresh terminal is required before the new command path becomes available.
Ask for help with finding uv after installation.
Your current shell may point to an older Python release. It may also have no python command yet.
- Check the current Python state by running this command:
python --versionWhat Does This Command Check?
The command asks the current shell which Python release it would run. The result determines how uv supplies the project’s managed 3.11 interpreter.
✔️ I see version 3.11 or higher
Your system already has a compatible release. Install uv’s managed Python 3.11 line to keep the project interpreter reproducible.
- Install the managed Python 3.11 line by running this command:
uv python install 3.11What Does This Command Do?
uv installs a managed interpreter from the Python 3.11 release line. The system interpreter remains separate.
The terminal should finish with Python 3.11 available to uv.
ⓧ I see an older version
Keep the older system release in place. uv can install the required Python 3.11 interpreter for this project.
- Install the managed Python 3.11 line by running this command:
uv python install 3.11What Does This Command Do?
uv installs the requested Python 3.11 line without replacing the older system release.
The terminal should finish with Python 3.11 available to uv.
ⓧ Command not found
uv can supply Python 3.11 even when the shell has no existing Python command.
- Install the managed Python 3.11 line by running this command:
uv python install 3.11What Does This Command Do?
uv downloads the requested Python 3.11 interpreter. The managed runtime becomes available for project environments.
The terminal should finish with Python 3.11 available to uv.
Python Installation Failed?
Confirm that uv reports version 0.13.6 in the current terminal. Retry after checking your internet connection.
Ask for targeted help with installing Python through uv.
- Create the project environment with Python 3.11 by running this command:
uv venv --python 3.11What Does This Command Do?
uv creates .venv inside the open project folder. The environment uses the managed Python 3.11 interpreter.
The terminal should confirm the new environment path. The .venv folder should appear in the Explorer.
The Environment Was Not Created?
Confirm that the integrated terminal is inside fail-closed-nse-data-pipeline. Check that the managed Python installation completed.
Ask for help with creating the project environment.
- Activate the project environment by running this command:
source .venv/bin/activateWhat Does This Command Do?
The command updates the current shell to use the interpreter inside .venv. Later package commands now target this project environment.
Your terminal prompt should now include .venv.
- Confirm the active interpreter by running this command:
python --versionWhat Does This Command Check?
The command now resolves through the active environment. Its output confirms which interpreter .venv provides.
The terminal should report Python 3.11 with a maintenance version after it.
Seeing a Different Python Version?
Confirm that the prompt includes .venv. Rerun the activation command from the project folder if the prompt does not show it.
Ask for help with activating the correct interpreter.
Create the reproducibility files
A dependency manifest gives uv exact package versions. A README.md file records the commands that recreate the workspace.
- Select New File in the Visual Studio Code Explorer.
- Enter requirements.txt as the file name.
- Add the pinned dependency manifest by pasting this content:
pandas==3.1.0
yfinance==1.7.0What Does This File Define?
- The pandas package stays on version 3.1.0 for predictable table handling.
- The yfinance package stays on version 1.7.0 for the research-data provider built later.
- Press Ctrl+S to save requirements.txt.
The Explorer should now list requirements.txt with both pinned package lines visible in the editor.
The Dependency File Looks Different?
Confirm that each package uses two equals signs before its version. Remove any spaces from the package lines.
Ask for help with checking the dependency manifest.
✔️ Awesome, I've got everything!
Great. Double-check that requirements.txt is saved before continuing.
ⓧ I'd like to double check the full code
pandas==3.1.0
yfinance==1.7.0- Synchronize the active environment with the manifest by running this command:
uv pip sync requirements.txtWhat Does This Command Do?
uv reads the exact versions from requirements.txt. It synchronizes the active environment to match that dependency list.
The terminal should finish with the pinned pandas and yfinance packages installed successfully.
The Dependency Sync Failed?
Confirm that the prompt still includes .venv. Check your internet connection before retrying the synchronization.
Ask for targeted help with synchronizing the pinned dependencies.
- Select New File in the Visual Studio Code Explorer.
- Enter README.md as the file name.
- Add the workspace overview plus setup commands by pasting this content:
# Fail-Closed NSE Data Pipeline
This is Stage 1 of a safety-first intraday trading system. It downloads research data, validates it, caches it, replays it, and refuses to approve yfinance for live trading.
It does not contain strategies, a broker adapter, or order placement.
## Linux setup
Verify Visual Studio Code:
```bash
code --version
```
Install uv 0.13.6 and create the Python 3.11 environment:
```bash
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | sh
uv python install 3.11
uv venv --python 3.11
source .venv/bin/activate
uv pip sync requirements.txt
```What Does This Section Record?
The opening defines the safety boundary for this stage. The setup section preserves the exact commands needed to recreate the Linux environment.
- Press Ctrl+S to save README.md.
The editor should show the project title plus the complete Linux setup section.
The Markdown Layout Looks Broken?
Confirm that every command group has an opening code fence plus a closing code fence. Keep one empty line around each fenced block.
Ask for help with checking the README setup section.
- Place the cursor on the first empty line below the setup code fence.
- Add the future pipeline commands by pasting this content:
## Run the pipeline
```bash
python data_pipeline.py fetch --symbol SBIN
python data_pipeline.py replay --symbol SBIN
python data_pipeline.py live-check
python -m unittest -v
```What Does This Section Record?
This section records the four commands the completed pipeline exposes. Later steps create the script plus tests behind those commands.
- Press Ctrl+S to save README.md.
The README should now end with a Run the pipeline section containing four command lines.
The New Section Is Missing?
Confirm that you added the block below the Linux setup code fence. Save the file after checking the heading starts on its own line.
Ask for help with placing the pipeline section correctly.
✔️ Awesome, I've got everything!
Great. Double-check that README.md is saved before the final environment check.
ⓧ I'd like to double check the full code
# Fail-Closed NSE Data Pipeline
This is Stage 1 of a safety-first intraday trading system. It downloads research data, validates it, caches it, replays it, and refuses to approve yfinance for live trading.
It does not contain strategies, a broker adapter, or order placement.
## Linux setup
Verify Visual Studio Code:
```bash
code --version
```
Install uv 0.13.6 and create the Python 3.11 environment:
```bash
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | sh
uv python install 3.11
uv venv --python 3.11
source .venv/bin/activate
uv pip sync requirements.txt
```
## Run the pipeline
```bash
python data_pipeline.py fetch --symbol SBIN
python data_pipeline.py replay --symbol SBIN
python data_pipeline.py live-check
python -m unittest -v
```
Before you run the final check, do you expect every command to resolve through the pinned workspace?
- Verify uv, Python, yfinance, and pandas by running this command:
uv --version && python --version && python -c "import yfinance, pandas; print(yfinance.__version__, pandas.__version__)"What Does This Check Prove?
- The uv check confirms the workspace tool release.
- The Python check confirms the interpreter selected by the active environment.
- The import check loads both pinned data packages from that environment.
The first line should report uv 0.13.6. The Python line should start with 3.11.
The final line should report 1.7.0 3.1.0. That is the reproducible foundation this pipeline needs.
A Version Does Not Match?
Confirm that the terminal prompt includes .venv. Rerun the dependency synchronization after checking both lines in requirements.txt.
Ask for help with diagnosing the version mismatch.
That foundation is now locked in. Your editor, managed interpreter, virtual environment, dependency pins, and setup guide all agree.
Your reproducible workspace is ready. Next, you will fetch and normalize your first NSE candles.
Fetch and Normalize NSE Candles
Your reproducible Python 3.11 workspace is ready. The next layer gives every future strategy one trusted candle shape.
The first yfinance download creates an early visible result. A provider abstraction lets you test the download lifecycle before later components depend on it.
In this step, get ready to:
- Define the research mode plus the approved NSE watchlist.
- Build a provider boundary for normalized OHLCV candles.
- Fetch recent five-minute SBIN candles through the validation layer.
Configure the research boundary
A TOML configuration file keeps provider settings outside the Python code. It also records the ₹50 hard daily-loss cap for later risk stages.
- Create config.toml inside the project folder using the file sidebar in Visual Studio Code.
- Add the research configuration by copying this code:
mode = "research"
provider = "yfinance"
interval = "5m"
period = "5d"
cache_path = "data/candles.sqlite3"
stale_after_seconds = 5
[risk]
max_daily_loss_rupees = 50
[watchlist]
HDFCBANK = "HDFCBANK.NS"
ICICIBANK = "ICICIBANK.NS"
SBIN = "SBIN.NS"
RELIANCE = "RELIANCE.NS"
INFY = "INFY.NS"
ITC = "ITC.NS"
TATASTEEL = "TATASTEEL.NS"What Does This Configuration Control?
- The mode value keeps this stage on the research path.
- The interval plus period values request recent five-minute candles.
- The stale_after_seconds value defines the future feed-health boundary.
- The watchlist table maps seven internal symbols to their Yahoo Finance symbols.
- Save config.toml.
- Confirm that the file ends with the TATASTEEL.NS mapping.
Configuration File Looks Different?
Check that every quoted value uses straight double quotes. Confirm that risk plus watchlist each have their own table header.
Ask for targeted help with checking the TOML structure.
Build the provider pipeline
A normalized candle gives the rest of the system one stable OHLCV shape. The provider boundary contains the vendor-specific download details.
Why Use a Provider Boundary?
Direct vendor calls would spread yfinance-specific behavior through future strategy code. A provider boundary gives that code one stable request plus one stable candle type.
A later provider can implement the same boundary. The validation rules remain in control.
- Create data_pipeline.py inside the project folder using the file sidebar in Visual Studio Code.
- Add the pipeline foundation by copying this code:
from __future__ import annotations
import argparse
import math
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
import yfinance as yf
class DataPipelineError(RuntimeError):
"""Base error for controlled data-pipeline failures."""
class DataUnavailable(DataPipelineError):
"""Raised when a provider cannot supply valid candles."""
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""
@dataclass(frozen=True)
class DataRequest:
internal_symbol: str
provider_symbol: str
interval: str
period: strWhat Does This Foundation Do?
- The imports provide command parsing plus numeric validation plus TOML loading.
- The controlled error hierarchy gives expected failures a shared base type.
- The frozen DataRequest records one provider download request.
- Save data_pipeline.py.
- Confirm that the final field is period: str.
Foundation Code Is Misaligned?
Check that each class begins at the left edge. Confirm that every field under DataRequest uses four spaces.
Ask for help with checking the pipeline foundation.
- Append the normalized candle shape to data_pipeline.py by copying this code:
@dataclass(frozen=True)
class Candle:
provider: str
internal_symbol: str
provider_symbol: str
interval: str
timestamp: str
open: float
high: float
low: float
close: float
volume: intWhat Does the Candle Store?
Each Candle keeps both symbol identities plus its interval plus its timestamp. The OHLCV fields preserve the values used by later analysis.
The frozen dataclass prevents accidental field changes after validation.
- Save data_pipeline.py.
- Confirm that Candle ends with the integer volume field.
Candle Fields Do Not Line Up?
Check that every field sits inside the Candle class. Keep the field names lowercase.
Ask for help with checking the Candle dataclass.
- Append the immutable settings shape to data_pipeline.py by copying this code:
@dataclass(frozen=True)
class Settings:
mode: str
provider: str
interval: str
period: str
cache_path: Path
stale_after_seconds: float
max_daily_loss_rupees: int
watchlist: dict[str, str]What Does Settings Represent?
The Settings type gives every required configuration value an explicit field. cache_path uses Path so later storage code receives a filesystem path.
- Save data_pipeline.py.
- Confirm that the final settings field is watchlist.
Settings Fields Are Missing?
Compare the settings fields with the keys in config.toml. Keep cache_path typed as Path.
Ask for help with matching Settings to the configuration.
- Append the configuration loader to data_pipeline.py by copying this first part:
def load_settings(path: Path) -> Settings:
with path.open("rb") as config_file:
raw = tomllib.load(config_file)
mode = str(raw["mode"]).lower()
if mode not in {"research", "paper", "live"}:
raise DataPipelineError(f"Unsupported mode: {mode}")
stale_after_seconds = float(raw["stale_after_seconds"])
if stale_after_seconds <= 0:
raise DataPipelineError("stale_after_seconds must be positive")
max_daily_loss_rupees = int(raw["risk"]["max_daily_loss_rupees"])
if max_daily_loss_rupees <= 0:
raise DataPipelineError("max_daily_loss_rupees must be positive")
watchlist = {
str(symbol).upper(): str(provider_symbol)
for symbol, provider_symbol in raw["watchlist"].items()
}
if not watchlist:
raise DataPipelineError("watchlist must not be empty")What Does the Loader Validate?
- The loader parses config.toml in binary mode.
- The mode check permits only the three planned operating modes.
- The numeric checks reject non-positive safety limits.
- The watchlist conversion standardizes internal symbols to uppercase.
- Save data_pipeline.py.
- Confirm that this part ends with the empty-watchlist check.
Loader Indentation Looks Wrong?
Keep each validation check inside load_settings(). Keep the watchlist comprehension aligned inside its braces.
Ask for help with checking the configuration loader.
- Complete load_settings() by adding this return block directly below the empty-watchlist check:
return Settings(
mode=mode,
provider=str(raw["provider"]).lower(),
interval=str(raw["interval"]),
period=str(raw["period"]),
cache_path=Path(str(raw["cache_path"])),
stale_after_seconds=stale_after_seconds,
max_daily_loss_rupees=max_daily_loss_rupees,
watchlist=watchlist,
)What Does This Return Block Do?
The return block converts the parsed values into one immutable Settings object. Downstream functions receive typed configuration instead of the raw TOML dictionary.
- Save data_pipeline.py.
- Confirm that the closing parenthesis aligns with return Settings(.
Return Block Sits Outside the Function?
Indent return Settings( with four spaces. Indent each supplied field with eight spaces.
Ask for help with placing the Settings return block.
- Append the provider boundary plus the yfinance download call to data_pipeline.py by copying this code:
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = False
def fetch_history(self, request: DataRequest) -> list[Candle]:
frame = yf.download(
tickers=request.provider_symbol,
period=request.period,
interval=request.interval,
actions=False,
threads=False,
auto_adjust=False,
progress=False,
timeout=10,
multi_level_index=False,
)How Does the Provider Boundary Work?
- The abstract method requires every provider to return normalized Candle objects.
- The live_approved capability starts as false.
- The download uses one-level columns for a single ticker.
- The raw-price option keeps OHLC values unadjusted for dividends.
- Save data_pipeline.py.
- Confirm that the download call ends with multi_level_index=False.
Download Call Is Nested Incorrectly?
Keep frame = yf.download( inside fetch_history(). Align each download option beneath the opening call.
Ask for help with checking the provider boundary.
- Extend fetch_history() with response checks by copying this code directly below the download call:
if frame is None or frame.empty:
raise DataUnavailable(
f"No candles returned for {request.provider_symbol}. "
"The symbol, Yahoo availability, network, or rate limit may be the cause."
)
required_columns = ["Open", "High", "Low", "Close", "Volume"]
missing = [column for column in required_columns if column not in frame.columns]
if missing:
raise DataUnavailable(f"Missing required columns: {missing}")
candles: list[Candle] = []
selected = frame[required_columns]
for row in selected.itertuples(index=True, name=None):
timestamp, open_price, high, low, close, volume = row
numeric_values = [
float(open_price),
float(high),
float(low),
float(close),
float(volume),
]
if not all(math.isfinite(value) for value in numeric_values):
raise DataUnavailable(
f"Non-finite candle received for {request.provider_symbol} at {timestamp}"
)How Are Responses Screened?
- An empty response becomes a controlled DataUnavailable failure.
- The column check requires all five OHLCV fields.
- The row iterator keeps the source timestamp beside its numeric values.
- The finite-number check rejects missing infinities plus invalid numeric values.
- Save data_pipeline.py.
- Confirm that this section ends inside the non-finite-value check.
Response Checks Fall Outside the Method?
Keep the response checks indented with eight spaces. Keep the row-processing lines indented inside the for block.
Ask for help with checking the response validation.
- Complete fetch_history() by copying this normalization block below the non-finite-value check:
candle = Candle(
provider=self.name,
internal_symbol=request.internal_symbol,
provider_symbol=request.provider_symbol,
interval=request.interval,
timestamp=timestamp.isoformat(),
open=numeric_values[0],
high=numeric_values[1],
low=numeric_values[2],
close=numeric_values[3],
volume=int(numeric_values[4]),
)
validate_candle(candle)
candles.append(candle)
if not candles:
raise DataUnavailable(f"No valid candles remained for {request.provider_symbol}")
return candlesHow Does Normalization Finish?
- Each source row becomes one frozen Candle.
- The timestamp converts to an ISO 8601 string.
- Every candle passes through validate_candle() before entering the result list.
- An empty normalized list becomes a controlled failure.
- Save data_pipeline.py.
- Confirm that fetch_history() ends with return candles.
Normalization Block Is Misaligned?
Keep the Candle creation inside the row loop. Align the final empty-list check with the for statement.
Ask for help with checking the normalization block.
- Append the candle rules to data_pipeline.py by copying this function:
def validate_candle(candle: Candle) -> None:
prices = [candle.open, candle.high, candle.low, candle.close]
if any(price <= 0 for price in prices):
raise DataUnavailable(f"Non-positive price at {candle.timestamp}")
if candle.volume < 0:
raise DataUnavailable(f"Negative volume at {candle.timestamp}")
if candle.low > candle.high:
raise DataUnavailable(f"Low exceeds high at {candle.timestamp}")
if not candle.low <= candle.open <= candle.high:
raise DataUnavailable(f"Open falls outside low/high at {candle.timestamp}")
if not candle.low <= candle.close <= candle.high:
raise DataUnavailable(f"Close falls outside low/high at {candle.timestamp}")What Makes a Candle Valid?
- Every OHLC price must be positive.
- Volume must be zero or higher.
- Low cannot exceed high.
- Open plus close must sit inside the low-to-high range.
- Save data_pipeline.py.
- Confirm that the final rule checks candle.close against the low-to-high range.
Validation Rules Raise Too Early?
Check the comparison directions for low plus high. Keep the non-positive test applied to all four prices.
Ask for help with checking the candle rules.
- Append the watchlist request builder to data_pipeline.py by copying this function:
def build_request(settings: Settings, internal_symbol: str) -> DataRequest:
symbol = internal_symbol.upper()
if symbol not in settings.watchlist:
raise DataPipelineError(f"Symbol {symbol} is outside the configured watchlist")
return DataRequest(
internal_symbol=symbol,
provider_symbol=settings.watchlist[symbol],
interval=settings.interval,
period=settings.period,
)What Does the Request Builder Protect?
The builder standardizes the requested internal symbol to uppercase. It refuses symbols outside the approved watchlist.
An approved symbol receives its configured provider mapping plus the shared interval plus the shared period.
- Save data_pipeline.py.
- Confirm that the function returns a DataRequest.
Watchlist Lookup Does Not Match?
Keep the uppercase conversion before the membership check. Confirm that config.toml uses uppercase internal symbols.
Ask for help with checking the request builder.
- Append the fetch execution function to data_pipeline.py by copying this code:
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
print(f"FETCHED={len(candles)}")
print(f"LAST_CANDLE={candles[-1]}")
return 0What Does Fetch Execution Prove?
The function permits the configured provider. It builds one approved request before downloading the candles.
The output reports the normalized batch size plus the newest validated candle.
- Save data_pipeline.py.
- Confirm that the function prints FETCHED plus LAST_CANDLE.
Fetch Execution References the Wrong Name?
Check that the provider class is named YFinanceProvider. Confirm that the request variable is passed to fetch_history().
Ask for help with checking run_fetch.
- Append the command parser to data_pipeline.py by copying this function:
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fail-closed NSE data pipeline")
parser.add_argument("--config", default="config.toml")
subparsers = parser.add_subparsers(dest="command", required=True)
fetch_parser = subparsers.add_parser("fetch")
fetch_parser.add_argument("--symbol", default="SBIN")
return parserWhat Does the Parser Accept?
The parser loads config.toml by default. The required fetch command accepts an approved internal symbol.
The default symbol is SBIN.
- Save data_pipeline.py.
- Confirm that build_parser() returns parser.
Parser Rejects the Fetch Command?
Check that fetch is added to subparsers. Keep the symbol flag attached to fetch_parser.
Ask for help with checking the command parser.
- Complete data_pipeline.py by copying this entry point:
def main() -> int:
args = build_parser().parse_args()
settings = load_settings(Path(args.config))
print(f"MODE={settings.mode}")
print(f"FUTURE_MAX_DAILY_LOSS_RUPEES={settings.max_daily_loss_rupees}")
try:
if args.command == "fetch":
return run_fetch(settings, args.symbol)
except DataPipelineError as error:
print(f"DATA PIPELINE HALTED: {error}")
return 1
raise DataPipelineError(f"Unknown command: {args.command}")
if __name__ == "__main__":
raise SystemExit(main())How Does the Entry Point Fail Closed?
The entry point loads the selected configuration before dispatching the command. It prints the active mode plus the reserved daily-loss limit.
A controlled pipeline error produces a halt message plus a non-zero result. An unknown command also stops execution.
- Save data_pipeline.py.
- Confirm that the file ends with raise SystemExit(main()).
Entry Point Does Not Run?
Check the two underscores on each side of __name__ plus __main__. Keep the fetch dispatch inside the try block.
Ask for help with checking the entry point.
✔️ Awesome, I've got everything!
Great. Double-check that config.toml plus data_pipeline.py are saved.
ⓧ I'd like to double check the full code
from __future__ import annotations
import argparse
import math
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
import yfinance as yf
class DataPipelineError(RuntimeError):
"""Base error for controlled data-pipeline failures."""
class DataUnavailable(DataPipelineError):
"""Raised when a provider cannot supply valid candles."""
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""
@dataclass(frozen=True)
class DataRequest:
internal_symbol: str
provider_symbol: str
interval: str
period: str
@dataclass(frozen=True)
class Candle:
provider: str
internal_symbol: str
provider_symbol: str
interval: str
timestamp: str
open: float
high: float
low: float
close: float
volume: int
@dataclass(frozen=True)
class Settings:
mode: str
provider: str
interval: str
period: str
cache_path: Path
stale_after_seconds: float
max_daily_loss_rupees: int
watchlist: dict[str, str]
def load_settings(path: Path) -> Settings:
with path.open("rb") as config_file:
raw = tomllib.load(config_file)
mode = str(raw["mode"]).lower()
if mode not in {"research", "paper", "live"}:
raise DataPipelineError(f"Unsupported mode: {mode}")
stale_after_seconds = float(raw["stale_after_seconds"])
if stale_after_seconds <= 0:
raise DataPipelineError("stale_after_seconds must be positive")
max_daily_loss_rupees = int(raw["risk"]["max_daily_loss_rupees"])
if max_daily_loss_rupees <= 0:
raise DataPipelineError("max_daily_loss_rupees must be positive")
watchlist = {
str(symbol).upper(): str(provider_symbol)
for symbol, provider_symbol in raw["watchlist"].items()
}
if not watchlist:
raise DataPipelineError("watchlist must not be empty")
return Settings(
mode=mode,
provider=str(raw["provider"]).lower(),
interval=str(raw["interval"]),
period=str(raw["period"]),
cache_path=Path(str(raw["cache_path"])),
stale_after_seconds=stale_after_seconds,
max_daily_loss_rupees=max_daily_loss_rupees,
watchlist=watchlist,
)
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = False
def fetch_history(self, request: DataRequest) -> list[Candle]:
frame = yf.download(
tickers=request.provider_symbol,
period=request.period,
interval=request.interval,
actions=False,
threads=False,
auto_adjust=False,
progress=False,
timeout=10,
multi_level_index=False,
)
if frame is None or frame.empty:
raise DataUnavailable(
f"No candles returned for {request.provider_symbol}. "
"The symbol, Yahoo availability, network, or rate limit may be the cause."
)
required_columns = ["Open", "High", "Low", "Close", "Volume"]
missing = [column for column in required_columns if column not in frame.columns]
if missing:
raise DataUnavailable(f"Missing required columns: {missing}")
candles: list[Candle] = []
selected = frame[required_columns]
for row in selected.itertuples(index=True, name=None):
timestamp, open_price, high, low, close, volume = row
numeric_values = [
float(open_price),
float(high),
float(low),
float(close),
float(volume),
]
if not all(math.isfinite(value) for value in numeric_values):
raise DataUnavailable(
f"Non-finite candle received for {request.provider_symbol} at {timestamp}"
)
candle = Candle(
provider=self.name,
internal_symbol=request.internal_symbol,
provider_symbol=request.provider_symbol,
interval=request.interval,
timestamp=timestamp.isoformat(),
open=numeric_values[0],
high=numeric_values[1],
low=numeric_values[2],
close=numeric_values[3],
volume=int(numeric_values[4]),
)
validate_candle(candle)
candles.append(candle)
if not candles:
raise DataUnavailable(f"No valid candles remained for {request.provider_symbol}")
return candles
def validate_candle(candle: Candle) -> None:
prices = [candle.open, candle.high, candle.low, candle.close]
if any(price <= 0 for price in prices):
raise DataUnavailable(f"Non-positive price at {candle.timestamp}")
if candle.volume < 0:
raise DataUnavailable(f"Negative volume at {candle.timestamp}")
if candle.low > candle.high:
raise DataUnavailable(f"Low exceeds high at {candle.timestamp}")
if not candle.low <= candle.open <= candle.high:
raise DataUnavailable(f"Open falls outside low/high at {candle.timestamp}")
if not candle.low <= candle.close <= candle.high:
raise DataUnavailable(f"Close falls outside low/high at {candle.timestamp}")
def build_request(settings: Settings, internal_symbol: str) -> DataRequest:
symbol = internal_symbol.upper()
if symbol not in settings.watchlist:
raise DataPipelineError(f"Symbol {symbol} is outside the configured watchlist")
return DataRequest(
internal_symbol=symbol,
provider_symbol=settings.watchlist[symbol],
interval=settings.interval,
period=settings.period,
)
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
print(f"FETCHED={len(candles)}")
print(f"LAST_CANDLE={candles[-1]}")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fail-closed NSE data pipeline")
parser.add_argument("--config", default="config.toml")
subparsers = parser.add_subparsers(dest="command", required=True)
fetch_parser = subparsers.add_parser("fetch")
fetch_parser.add_argument("--symbol", default="SBIN")
return parser
def main() -> int:
args = build_parser().parse_args()
settings = load_settings(Path(args.config))
print(f"MODE={settings.mode}")
print(f"FUTURE_MAX_DAILY_LOSS_RUPEES={settings.max_daily_loss_rupees}")
try:
if args.command == "fetch":
return run_fetch(settings, args.symbol)
except DataPipelineError as error:
print(f"DATA PIPELINE HALTED: {error}")
return 1
raise DataPipelineError(f"Unknown command: {args.command}")
if __name__ == "__main__":
raise SystemExit(main())Fetch and inspect SBIN candles
The complete path now turns a configured internal symbol into validated provider candles. The final check reveals whether the market timestamp survives normalization.
Before you run this check, which details should prove that the latest candle kept its provider identity plus its timezone?
- Fetch recent SBIN candles through the pipeline by running this command:
python data_pipeline.py fetch --symbol SBINWhat Does This Command Test?
The command loads the research configuration. It maps SBIN to SBIN.NS before calling the provider.
Every returned row passes through numeric checks plus OHLC relationship checks before the newest candle is printed.
You should see a positive FETCHED count. The LAST_CANDLE output should name SBIN.NS plus a timestamp with a timezone offset.
Download Did Not Return Candles?
Confirm that the activated environment still provides yfinance 1.7.0. Check your internet connection before retrying.
The provider can also return no data when Yahoo availability or rate limits interfere. Ask for help with diagnosing the controlled download failure.
Before you inspect the project files, do you think the downloaded candles still exist after the command exits?
- Check the current project files using the file sidebar in Visual Studio Code.
You should still see config.toml plus data_pipeline.py. You should not see the configured data folder or a candle database.
Why Did the Candles Disappear?
The provider returned valid candles to the running Python process. The process printed the newest object before releasing the in-memory list when it exited.
This shortfall is intentional. You have proved that validation works before adding persistent storage.
That is your first trusted market-data boundary working. Next, you will preserve each validated candle without creating duplicates.
Cache Candles Idempotently
Your yfinance provider now turns recent SBIN data into validated Candle objects. Those objects vanish when the fetch process exits.
SQLite gives every candle a durable identity. This step makes repeated downloads safe by keeping one row for each provider, symbol, interval, and timestamp.
In this step, get ready to:
- Create a local candle table with a composite primary key.
- Store validated candle batches with parameterized SQL.
- Prove that repeated writes cannot duplicate cached candles.
Create the cache schema
A composite primary key combines several fields into one unique identity. The candle table uses the provider, internal symbol, interval, and timestamp to identify each market observation.
- In the Visual Studio Code Explorer sidebar, select data_pipeline.py.
- Replace the existing import block with the following imports:
from __future__ import annotations
import argparse
import math
import sqlite3
import time
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import yfinance as yfWhat Do These Imports Add?
- The sqlite3 module creates the local database without another dependency.
- The datetime tools record when each batch reaches the cache.
- The time module prepares the pipeline for its later feed-safety checks.
- Find the line that starts def build_request.
- Add the cache class directly above that line by pasting this code:
class CandleCache:
def __init__(self, path: Path):
self.path = path
def initialize(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(self.path)
try:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS candles (
provider TEXT NOT NULL,
internal_symbol TEXT NOT NULL,
provider_symbol TEXT NOT NULL,
interval TEXT NOT NULL,
timestamp TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume INTEGER NOT NULL,
downloaded_at TEXT NOT NULL,
PRIMARY KEY (provider, internal_symbol, interval, timestamp)
)
"""
)
connection.commit()
finally:
connection.close()What Does This Code Do?
- The constructor remembers the database path from config.toml.
- The initialize() method creates the parent data directory when needed.
- The table stores normalized OHLCV values plus the batch download time.
- The primary key prevents two rows from representing the same candle identity.
- Save data_pipeline.py.
- Confirm that the pipeline still fetches validated candles by running:
python data_pipeline.py fetch --symbol SBINWhat Should You See?
The terminal should print a positive FETCHED count. It should also print a timezone-aware LAST_CANDLE for SBIN.NS.
Seeing a Python Error?
Check that CandleCache sits above build_request(). Confirm that every line inside the class keeps its indentation.
Ask for help with diagnosing the cache schema code.
Add idempotent storage and reads
An idempotent write can be retried without changing the result after its first success. The cache uses INSERT OR IGNORE so a primary-key conflict leaves the existing candle untouched.
- Place the first part of store() below the initialize() method:
def store(self, candles: list[Candle]) -> int:
if not candles:
return 0
downloaded_at = datetime.now(timezone.utc).isoformat()
rows = [
(
candle.provider,
candle.internal_symbol,
candle.provider_symbol,
candle.interval,
candle.timestamp,
candle.open,
candle.high,
candle.low,
candle.close,
candle.volume,
downloaded_at,
)
for candle in candles
]How Are the Rows Prepared?
- An empty batch returns zero without touching the database.
- The batch receives one UTC download timestamp.
- Each immutable Candle becomes a tuple that matches the table column order.
- Continue the same store() method directly below the rows list:
connection = sqlite3.connect(self.path)
try:
before = connection.total_changes
connection.executemany(
"""
INSERT OR IGNORE INTO candles (
provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume, downloaded_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
connection.commit()
return connection.total_changes - before
finally:
connection.close()How Does the Write Stay Safe?
- The question-mark placeholders keep candle values separate from the SQL statement.
- The executemany() call applies the same parameterized statement to the whole batch.
- The change counter reports how many new rows SQLite accepted.
- The finally block closes the connection after success or failure.
- Save data_pipeline.py.
- Check that the completed storage method preserves the working fetch by running:
python data_pipeline.py fetch --symbol SBINWhat Should You See?
The terminal should still print validated SBIN candles. This confirms that the new storage method parses cleanly before it joins the fetch path.
- Add the count() method directly below store() by pasting this code:
def count(self, provider: str, internal_symbol: str, interval: str) -> int:
connection = sqlite3.connect(self.path)
try:
result = connection.execute(
"""
SELECT COUNT(*)
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
""",
(provider, internal_symbol, interval),
).fetchone()
return int(result[0])
finally:
connection.close()What Does the Count Measure?
- The query counts cached rows for one provider, internal symbol, and interval.
- Parameterized filters ensure the count matches the same identity fields used by the fetch.
- Add the load() method directly below count() by pasting this code:
def load(self, provider: str, internal_symbol: str, interval: str) -> list[Candle]:
connection = sqlite3.connect(self.path)
try:
rows = connection.execute(
"""
SELECT provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
ORDER BY timestamp ASC
""",
(provider, internal_symbol, interval),
).fetchall()
finally:
connection.close()
return [Candle(*row) for row in rows]Why Load in Timestamp Order?
- The query rebuilds the same normalized Candle objects used by the provider boundary.
- The ascending timestamp order makes later replay deterministic.
- The selected column order matches the Candle dataclass field order.
- Save data_pipeline.py.
- Confirm that every cache method parses by running the fetch command again:
python data_pipeline.py fetch --symbol SBINWhat Should You See?
The terminal should complete the fetch without a syntax error. Your cache now has write, count, and ordered-read operations ready for the fetch path.
Cache Methods Not Parsing?
Keep store(), count(), and load() aligned at the same indentation level inside CandleCache.
Ask for help with checking the CandleCache method structure.
Connect the cache to fetch
The fetch path can now turn a transient download into persistent state. Writing the same batch twice creates an immediate test of the uniqueness rule.
- Find the existing run_fetch() function in data_pipeline.py.
- Replace the entire function with this cache-enabled version:
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
cache = CandleCache(settings.cache_path)
cache.initialize()
first_inserted = cache.store(candles)
duplicate_inserted = cache.store(candles)
cached_rows = cache.count(provider.name, request.internal_symbol, request.interval)
print(f"FETCHED={len(candles)}")
print(f"FIRST_INSERTED={first_inserted}")
print(f"DUPLICATE_INSERTED={duplicate_inserted}")
print(f"CACHE_ROWS={cached_rows}")
print(f"LAST_CANDLE={candles[-1]}")
return 0How Does Fetch Test Idempotency?
- The cache initializes itself at the path from config.toml.
- The first store() call inserts candle identities that are absent.
- The second store() call retries the identical batch.
- The printed counters expose the database result without requiring a separate database tool.
- Save data_pipeline.py.
Before you run the cache-enabled fetch, which insertion counter should reveal that the repeated write was harmless?
- Create the database and test the duplicate write by running:
python data_pipeline.py fetch --symbol SBINWhat Should You See?
The terminal prints a positive FETCHED count. It also prints DUPLICATE_INSERTED=0 and a positive CACHE_ROWS value.
You now have data/candles.sqlite3 with one canonical row per candle identity.
Database Was Not Created?
Confirm that cache_path in config.toml is data/candles.sqlite3. Check that cache.initialize() appears before the first storage call.
Ask for help with diagnosing the missing SQLite cache.
Before you rerun the same fetch, do you expect the duplicate insertion count to change?
- Verify the persistent cache with the same fetch command:
python data_pipeline.py fetch --symbol SBINWhat Proves the Retry Is Safe?
The terminal prints DUPLICATE_INSERTED=0 again. The CACHE_ROWS value remains positive because the earlier candles persist on disk.
That is the retry guarantee working. Repeating a download can preserve existing candles without creating duplicate records.
✔️ Awesome, I've got everything
Great work. Save data_pipeline.py before continuing.
ⓧ I'd like to double check the full code
from __future__ import annotations
import argparse
import math
import sqlite3
import time
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import yfinance as yf
class DataPipelineError(RuntimeError):
"""Base error for controlled data-pipeline failures."""
class DataUnavailable(DataPipelineError):
"""Raised when a provider cannot supply valid candles."""
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""
@dataclass(frozen=True)
class DataRequest:
internal_symbol: str
provider_symbol: str
interval: str
period: str
@dataclass(frozen=True)
class Candle:
provider: str
internal_symbol: str
provider_symbol: str
interval: str
timestamp: str
open: float
high: float
low: float
close: float
volume: int
@dataclass(frozen=True)
class Settings:
mode: str
provider: str
interval: str
period: str
cache_path: Path
stale_after_seconds: float
max_daily_loss_rupees: int
watchlist: dict[str, str]
def load_settings(path: Path) -> Settings:
with path.open("rb") as config_file:
raw = tomllib.load(config_file)
mode = str(raw["mode"]).lower()
if mode not in {"research", "paper", "live"}:
raise DataPipelineError(f"Unsupported mode: {mode}")
stale_after_seconds = float(raw["stale_after_seconds"])
if stale_after_seconds <= 0:
raise DataPipelineError("stale_after_seconds must be positive")
max_daily_loss_rupees = int(raw["risk"]["max_daily_loss_rupees"])
if max_daily_loss_rupees <= 0:
raise DataPipelineError("max_daily_loss_rupees must be positive")
watchlist = {
str(symbol).upper(): str(provider_symbol)
for symbol, provider_symbol in raw["watchlist"].items()
}
if not watchlist:
raise DataPipelineError("watchlist must not be empty")
return Settings(
mode=mode,
provider=str(raw["provider"]).lower(),
interval=str(raw["interval"]),
period=str(raw["period"]),
cache_path=Path(str(raw["cache_path"])),
stale_after_seconds=stale_after_seconds,
max_daily_loss_rupees=max_daily_loss_rupees,
watchlist=watchlist,
)
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = False
def fetch_history(self, request: DataRequest) -> list[Candle]:
frame = yf.download(
tickers=request.provider_symbol,
period=request.period,
interval=request.interval,
actions=False,
threads=False,
auto_adjust=False,
progress=False,
timeout=10,
multi_level_index=False,
)
if frame is None or frame.empty:
raise DataUnavailable(
f"No candles returned for {request.provider_symbol}. "
"The symbol, Yahoo availability, network, or rate limit may be the cause."
)
required_columns = ["Open", "High", "Low", "Close", "Volume"]
missing = [column for column in required_columns if column not in frame.columns]
if missing:
raise DataUnavailable(f"Missing required columns: {missing}")
candles: list[Candle] = []
selected = frame[required_columns]
for row in selected.itertuples(index=True, name=None):
timestamp, open_price, high, low, close, volume = row
numeric_values = [
float(open_price),
float(high),
float(low),
float(close),
float(volume),
]
if not all(math.isfinite(value) for value in numeric_values):
raise DataUnavailable(
f"Non-finite candle received for {request.provider_symbol} at {timestamp}"
)
candle = Candle(
provider=self.name,
internal_symbol=request.internal_symbol,
provider_symbol=request.provider_symbol,
interval=request.interval,
timestamp=timestamp.isoformat(),
open=numeric_values[0],
high=numeric_values[1],
low=numeric_values[2],
close=numeric_values[3],
volume=int(numeric_values[4]),
)
validate_candle(candle)
candles.append(candle)
if not candles:
raise DataUnavailable(f"No valid candles remained for {request.provider_symbol}")
return candles
def validate_candle(candle: Candle) -> None:
prices = [candle.open, candle.high, candle.low, candle.close]
if any(price <= 0 for price in prices):
raise DataUnavailable(f"Non-positive price at {candle.timestamp}")
if candle.volume < 0:
raise DataUnavailable(f"Negative volume at {candle.timestamp}")
if candle.low > candle.high:
raise DataUnavailable(f"Low exceeds high at {candle.timestamp}")
if not candle.low <= candle.open <= candle.high:
raise DataUnavailable(f"Open falls outside low/high at {candle.timestamp}")
if not candle.low <= candle.close <= candle.high:
raise DataUnavailable(f"Close falls outside low/high at {candle.timestamp}")
class CandleCache:
def __init__(self, path: Path):
self.path = path
def initialize(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(self.path)
try:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS candles (
provider TEXT NOT NULL,
internal_symbol TEXT NOT NULL,
provider_symbol TEXT NOT NULL,
interval TEXT NOT NULL,
timestamp TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume INTEGER NOT NULL,
downloaded_at TEXT NOT NULL,
PRIMARY KEY (provider, internal_symbol, interval, timestamp)
)
"""
)
connection.commit()
finally:
connection.close()
def store(self, candles: list[Candle]) -> int:
if not candles:
return 0
downloaded_at = datetime.now(timezone.utc).isoformat()
rows = [
(
candle.provider,
candle.internal_symbol,
candle.provider_symbol,
candle.interval,
candle.timestamp,
candle.open,
candle.high,
candle.low,
candle.close,
candle.volume,
downloaded_at,
)
for candle in candles
]
connection = sqlite3.connect(self.path)
try:
before = connection.total_changes
connection.executemany(
"""
INSERT OR IGNORE INTO candles (
provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume, downloaded_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
connection.commit()
return connection.total_changes - before
finally:
connection.close()
def count(self, provider: str, internal_symbol: str, interval: str) -> int:
connection = sqlite3.connect(self.path)
try:
result = connection.execute(
"""
SELECT COUNT(*)
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
""",
(provider, internal_symbol, interval),
).fetchone()
return int(result[0])
finally:
connection.close()
def load(self, provider: str, internal_symbol: str, interval: str) -> list[Candle]:
connection = sqlite3.connect(self.path)
try:
rows = connection.execute(
"""
SELECT provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
ORDER BY timestamp ASC
""",
(provider, internal_symbol, interval),
).fetchall()
finally:
connection.close()
return [Candle(*row) for row in rows]
def build_request(settings: Settings, internal_symbol: str) -> DataRequest:
symbol = internal_symbol.upper()
if symbol not in settings.watchlist:
raise DataPipelineError(f"Symbol {symbol} is outside the configured watchlist")
return DataRequest(
internal_symbol=symbol,
provider_symbol=settings.watchlist[symbol],
interval=settings.interval,
period=settings.period,
)
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
cache = CandleCache(settings.cache_path)
cache.initialize()
first_inserted = cache.store(candles)
duplicate_inserted = cache.store(candles)
cached_rows = cache.count(provider.name, request.internal_symbol, request.interval)
print(f"FETCHED={len(candles)}")
print(f"FIRST_INSERTED={first_inserted}")
print(f"DUPLICATE_INSERTED={duplicate_inserted}")
print(f"CACHE_ROWS={cached_rows}")
print(f"LAST_CANDLE={candles[-1]}")
return 0
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fail-closed NSE data pipeline")
parser.add_argument("--config", default="config.toml")
subparsers = parser.add_subparsers(dest="command", required=True)
fetch_parser = subparsers.add_parser("fetch")
fetch_parser.add_argument("--symbol", default="SBIN")
return parser
def main() -> int:
args = build_parser().parse_args()
settings = load_settings(Path(args.config))
print(f"MODE={settings.mode}")
print(f"FUTURE_MAX_DAILY_LOSS_RUPEES={settings.max_daily_loss_rupees}")
try:
if args.command == "fetch":
return run_fetch(settings, args.symbol)
except DataPipelineError as error:
print(f"DATA PIPELINE HALTED: {error}")
return 1
raise DataPipelineError(f"Unknown command: {args.command}")
if __name__ == "__main__":
raise SystemExit(main())Your validated candles now survive process exits without multiplying during retries. Next, you will add a permission gate that blocks research data from any future live-order path.
Block Unsafe Live Data
Your SQLite cache now preserves validated candles across runs. A successful yfinance download proves that candles arrived.
Live safety requires a separate permission decision because yfinance provides no NSE latency or service guarantee. You will add a gate that checks explicit provider approval plus message age before future broker code can receive live data.
In this step, get ready to:
- Add explicit fail-closed capabilities to every data provider.
- Enforce provider approval plus feed freshness through LiveDataGate.
- Run a controlled live-data check against yfinance.
Declare provider permissions
A provider capability records whether a data source has permission to support live decisions. The default value stays false so every new provider begins in the safest state.
- In data_pipeline.py, locate the existing DataProvider class.
- Replace the complete class with this version:
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""Why Default to False?
The base class gives every provider an explicit name plus a live-data capability. A newly added provider remains blocked until its implementation deliberately changes live_approved.
- In YFinanceProvider, find the line name = "yfinance".
- Replace the class header with this explicit research-only configuration:
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = FalseWhat Does This Capability Mean?
A successful download does not change this permission. YFinanceProvider remains blocked from live decisions even when its candles pass every validation rule.
- Save data_pipeline.py.
- Confirm that the existing fetch path still works by running:
python data_pipeline.py fetch --symbol SBINWhat Does This Check Prove?
The command still downloads normalized SBIN.NS candles through the provider boundary. It also confirms that the new class capabilities leave research-mode fetching intact.
You should see a positive FETCHED count. You should also see DUPLICATE_INSERTED=0 with a positive CACHE_ROWS count.
Fetch Stopped Working?
Check that fetch_history() remains indented inside YFinanceProvider. Confirm that both capability lines sit above that method.
Ask for targeted help with repairing the provider class structure.
Enforce live-feed readiness
The safety gate checks two independent conditions. It requires an approved provider plus a recent message measured by a monotonic clock.
- In data_pipeline.py, locate the existing DataUnavailable class.
- Add this controlled live-data exception immediately below it:
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""Why Use a Controlled Exception?
Live-data refusal is an expected safety outcome. LiveDataBlocked lets the command report that outcome without hiding unrelated programming failures.
- Find the existing build_request() function below CandleCache.
- Add the complete safety gate immediately above build_request() by pasting:
class LiveDataGate:
def __init__(self, stale_after_seconds: float):
if stale_after_seconds <= 0:
raise ValueError("stale_after_seconds must be positive")
self.stale_after_seconds = stale_after_seconds
def require_ready(
self,
provider: DataProvider,
mode: str,
last_message_monotonic: float | None,
now_monotonic: float | None = None,
) -> None:
if mode != "live":
return
if not provider.live_approved:
raise LiveDataBlocked(
f"Provider {provider.name} is not approved for live trading"
)
if last_message_monotonic is None:
raise LiveDataBlocked("No live-feed message has been received")
now = time.monotonic() if now_monotonic is None else now_monotonic
age = now - last_message_monotonic
if age < 0:
raise LiveDataBlocked("Monotonic feed clock moved backwards")
if age > self.stale_after_seconds:
raise LiveDataBlocked(
f"Feed age {age:.1f}s exceeds {self.stale_after_seconds:.1f}s"
)How Does the Gate Fail Closed?
- The mode check leaves research and paper workflows available.
- The capability check blocks every provider that has not received explicit live approval.
- The message check blocks feeds with no observed message timestamp.
- The age checks reject a backward clock or a feed older than stale_after_seconds.
- Save data_pipeline.py.
- Confirm that research-mode fetching remains available by running:
python data_pipeline.py fetch --symbol SBINWhy Does Fetch Still Work?
The gate performs enforcement only when its caller requests live mode. The configured research workflow continues to fetch plus cache validated candles.
You should see another positive FETCHED count. The cached batch should still report DUPLICATE_INSERTED=0.
Seeing a Gate Syntax Error?
Check that the complete LiveDataGate class sits at the left margin. Confirm that build_request() begins only after the class ends.
Ask for help with checking the LiveDataGate indentation.
Wire the controlled live check
The command path deliberately presents the research provider to the live gate. It catches only the expected safety exception so unexpected failures remain visible.
- Find the end of the existing run_fetch() function in data_pipeline.py.
- Add this live-check function below run_fetch() by pasting:
def run_live_check(settings: Settings) -> int:
gate = LiveDataGate(settings.stale_after_seconds)
provider = YFinanceProvider()
try:
gate.require_ready(
provider=provider,
mode="live",
last_message_monotonic=None,
)
except LiveDataBlocked as error:
print(f"LIVE BLOCKED: {error}")
return 0
raise DataPipelineError("Unsafe result: yfinance unexpectedly passed the live gate")What Does This Function Protect?
The function creates the gate from the configured stale threshold. It then asks YFinanceProvider to pass a live-mode readiness check.
The expected refusal becomes a controlled terminal message. An unexpected approval raises DataPipelineError because continuing would violate the safety boundary.
- Inside build_parser(), locate the final return parser line.
- Register the new command immediately above that return line by adding:
subparsers.add_parser("live-check")What Does This Parser Line Do?
The parser now recognizes live-check as a valid command. The command needs no symbol because it tests provider permission instead of downloading candles.
- Save data_pipeline.py.
- Confirm that the existing fetch command still reaches its original handler by running:
python data_pipeline.py fetch --symbol SBINWhat Does This Regression Check Cover?
The command confirms that adding a second parser path did not change the existing fetch path. Your normalized download plus idempotent cache should still complete.
You should see MODE=research followed by the existing fetch and cache results.
Fetch Command Became Invalid?
Confirm that subparsers.add_parser("live-check") uses the same indentation as the existing fetch parser lines. Keep it above return parser.
Ask for help with repairing the command parser.
- Inside main(), locate the existing if args.command == "fetch": branch.
- Add the live-check branch immediately below the fetch branch by pasting:
if args.command == "live-check":
return run_live_check(settings)How Is the Command Dispatched?
The new branch sends live-check to run_live_check(). The surrounding error handler continues to catch unexpected pipeline failures.
- Save data_pipeline.py.
Before you run the check, do you expect a successful research download to grant this provider live permission?
- Test the live-data boundary by running:
python data_pipeline.py live-checkWhat Does This Command Test?
The command requests live-mode approval for the configured yfinance provider. The gate evaluates its explicit capability before checking message freshness.
You should see LIVE BLOCKED: Provider yfinance is not approved for live trading. The program reports a controlled refusal instead of crashing or continuing.
Did the Live Check Fail Differently?
Confirm that YFinanceProvider.live_approved remains False. Check that run_live_check() passes mode="live" to the gate.
Ask for help with diagnosing the controlled live-data refusal.
That is the safety boundary working: research candles remain useful while the same provider stays outside the future live-order path.
✔️ Awesome, I've got everything!
Great. Double-check that you saved data_pipeline.py with the controlled live-data gate.
ⓧ I'd like to double check the full code
from __future__ import annotations
import argparse
import math
import sqlite3
import time
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import yfinance as yf
class DataPipelineError(RuntimeError):
"""Base error for controlled data-pipeline failures."""
class DataUnavailable(DataPipelineError):
"""Raised when a provider cannot supply valid candles."""
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""
@dataclass(frozen=True)
class DataRequest:
internal_symbol: str
provider_symbol: str
interval: str
period: str
@dataclass(frozen=True)
class Candle:
provider: str
internal_symbol: str
provider_symbol: str
interval: str
timestamp: str
open: float
high: float
low: float
close: float
volume: int
@dataclass(frozen=True)
class Settings:
mode: str
provider: str
interval: str
period: str
cache_path: Path
stale_after_seconds: float
max_daily_loss_rupees: int
watchlist: dict[str, str]
def load_settings(path: Path) -> Settings:
with path.open("rb") as config_file:
raw = tomllib.load(config_file)
mode = str(raw["mode"]).lower()
if mode not in {"research", "paper", "live"}:
raise DataPipelineError(f"Unsupported mode: {mode}")
stale_after_seconds = float(raw["stale_after_seconds"])
if stale_after_seconds <= 0:
raise DataPipelineError("stale_after_seconds must be positive")
max_daily_loss_rupees = int(raw["risk"]["max_daily_loss_rupees"])
if max_daily_loss_rupees <= 0:
raise DataPipelineError("max_daily_loss_rupees must be positive")
watchlist = {
str(symbol).upper(): str(provider_symbol)
for symbol, provider_symbol in raw["watchlist"].items()
}
if not watchlist:
raise DataPipelineError("watchlist must not be empty")
return Settings(
mode=mode,
provider=str(raw["provider"]).lower(),
interval=str(raw["interval"]),
period=str(raw["period"]),
cache_path=Path(str(raw["cache_path"])),
stale_after_seconds=stale_after_seconds,
max_daily_loss_rupees=max_daily_loss_rupees,
watchlist=watchlist,
)
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = False
def fetch_history(self, request: DataRequest) -> list[Candle]:
frame = yf.download(
tickers=request.provider_symbol,
period=request.period,
interval=request.interval,
actions=False,
threads=False,
auto_adjust=False,
progress=False,
timeout=10,
multi_level_index=False,
)
if frame is None or frame.empty:
raise DataUnavailable(
f"No candles returned for {request.provider_symbol}. "
"The symbol, Yahoo availability, network, or rate limit may be the cause."
)
required_columns = ["Open", "High", "Low", "Close", "Volume"]
missing = [column for column in required_columns if column not in frame.columns]
if missing:
raise DataUnavailable(f"Missing required columns: {missing}")
candles: list[Candle] = []
selected = frame[required_columns]
for row in selected.itertuples(index=True, name=None):
timestamp, open_price, high, low, close, volume = row
numeric_values = [
float(open_price),
float(high),
float(low),
float(close),
float(volume),
]
if not all(math.isfinite(value) for value in numeric_values):
raise DataUnavailable(
f"Non-finite candle received for {request.provider_symbol} at {timestamp}"
)
candle = Candle(
provider=self.name,
internal_symbol=request.internal_symbol,
provider_symbol=request.provider_symbol,
interval=request.interval,
timestamp=timestamp.isoformat(),
open=numeric_values[0],
high=numeric_values[1],
low=numeric_values[2],
close=numeric_values[3],
volume=int(numeric_values[4]),
)
validate_candle(candle)
candles.append(candle)
if not candles:
raise DataUnavailable(f"No valid candles remained for {request.provider_symbol}")
return candles
def validate_candle(candle: Candle) -> None:
prices = [candle.open, candle.high, candle.low, candle.close]
if any(price <= 0 for price in prices):
raise DataUnavailable(f"Non-positive price at {candle.timestamp}")
if candle.volume < 0:
raise DataUnavailable(f"Negative volume at {candle.timestamp}")
if candle.low > candle.high:
raise DataUnavailable(f"Low exceeds high at {candle.timestamp}")
if not candle.low <= candle.open <= candle.high:
raise DataUnavailable(f"Open falls outside low/high at {candle.timestamp}")
if not candle.low <= candle.close <= candle.high:
raise DataUnavailable(f"Close falls outside low/high at {candle.timestamp}")
class CandleCache:
def __init__(self, path: Path):
self.path = path
def initialize(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(self.path)
try:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS candles (
provider TEXT NOT NULL,
internal_symbol TEXT NOT NULL,
provider_symbol TEXT NOT NULL,
interval TEXT NOT NULL,
timestamp TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume INTEGER NOT NULL,
downloaded_at TEXT NOT NULL,
PRIMARY KEY (provider, internal_symbol, interval, timestamp)
)
"""
)
connection.commit()
finally:
connection.close()
def store(self, candles: list[Candle]) -> int:
if not candles:
return 0
downloaded_at = datetime.now(timezone.utc).isoformat()
rows = [
(
candle.provider,
candle.internal_symbol,
candle.provider_symbol,
candle.interval,
candle.timestamp,
candle.open,
candle.high,
candle.low,
candle.close,
candle.volume,
downloaded_at,
)
for candle in candles
]
connection = sqlite3.connect(self.path)
try:
before = connection.total_changes
connection.executemany(
"""
INSERT OR IGNORE INTO candles (
provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume, downloaded_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
connection.commit()
return connection.total_changes - before
finally:
connection.close()
def count(self, provider: str, internal_symbol: str, interval: str) -> int:
connection = sqlite3.connect(self.path)
try:
result = connection.execute(
"""
SELECT COUNT(*)
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
""",
(provider, internal_symbol, interval),
).fetchone()
return int(result[0])
finally:
connection.close()
def load(self, provider: str, internal_symbol: str, interval: str) -> list[Candle]:
connection = sqlite3.connect(self.path)
try:
rows = connection.execute(
"""
SELECT provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
ORDER BY timestamp ASC
""",
(provider, internal_symbol, interval),
).fetchall()
finally:
connection.close()
return [Candle(*row) for row in rows]
class LiveDataGate:
def __init__(self, stale_after_seconds: float):
if stale_after_seconds <= 0:
raise ValueError("stale_after_seconds must be positive")
self.stale_after_seconds = stale_after_seconds
def require_ready(
self,
provider: DataProvider,
mode: str,
last_message_monotonic: float | None,
now_monotonic: float | None = None,
) -> None:
if mode != "live":
return
if not provider.live_approved:
raise LiveDataBlocked(
f"Provider {provider.name} is not approved for live trading"
)
if last_message_monotonic is None:
raise LiveDataBlocked("No live-feed message has been received")
now = time.monotonic() if now_monotonic is None else now_monotonic
age = now - last_message_monotonic
if age < 0:
raise LiveDataBlocked("Monotonic feed clock moved backwards")
if age > self.stale_after_seconds:
raise LiveDataBlocked(
f"Feed age {age:.1f}s exceeds {self.stale_after_seconds:.1f}s"
)
def build_request(settings: Settings, internal_symbol: str) -> DataRequest:
symbol = internal_symbol.upper()
if symbol not in settings.watchlist:
raise DataPipelineError(f"Symbol {symbol} is outside the configured watchlist")
return DataRequest(
internal_symbol=symbol,
provider_symbol=settings.watchlist[symbol],
interval=settings.interval,
period=settings.period,
)
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
cache = CandleCache(settings.cache_path)
cache.initialize()
first_inserted = cache.store(candles)
duplicate_inserted = cache.store(candles)
cached_rows = cache.count(provider.name, request.internal_symbol, request.interval)
print(f"FETCHED={len(candles)}")
print(f"FIRST_INSERTED={first_inserted}")
print(f"DUPLICATE_INSERTED={duplicate_inserted}")
print(f"CACHE_ROWS={cached_rows}")
print(f"LAST_CANDLE={candles[-1]}")
return 0
def run_live_check(settings: Settings) -> int:
gate = LiveDataGate(settings.stale_after_seconds)
provider = YFinanceProvider()
try:
gate.require_ready(
provider=provider,
mode="live",
last_message_monotonic=None,
)
except LiveDataBlocked as error:
print(f"LIVE BLOCKED: {error}")
return 0
raise DataPipelineError("Unsafe result: yfinance unexpectedly passed the live gate")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fail-closed NSE data pipeline")
parser.add_argument("--config", default="config.toml")
subparsers = parser.add_subparsers(dest="command", required=True)
fetch_parser = subparsers.add_parser("fetch")
fetch_parser.add_argument("--symbol", default="SBIN")
subparsers.add_parser("live-check")
return parser
def main() -> int:
args = build_parser().parse_args()
settings = load_settings(Path(args.config))
print(f"MODE={settings.mode}")
print(f"FUTURE_MAX_DAILY_LOSS_RUPEES={settings.max_daily_loss_rupees}")
try:
if args.command == "fetch":
return run_fetch(settings, args.symbol)
if args.command == "live-check":
return run_live_check(settings)
except DataPipelineError as error:
print(f"DATA PIPELINE HALTED: {error}")
return 1
raise DataPipelineError(f"Unknown command: {args.command}")
if __name__ == "__main__":
raise SystemExit(main())Your pipeline now preserves useful research data without granting it live authority. Next, you will replay cached candles deterministically and prove these safety rules with offline tests.
Replay Data and Test Safety Rules
Your SQLite cache already preserves normalized candles. Your live gate already blocks yfinance from future order decisions.
Backtests need deterministic replay so every run receives candles in the same order. Offline unit tests protect that ordering plus the fail-closed rules during later refactors.
In this step, get ready to:
- Replay normalized candles from the cache in timestamp order.
- Test duplicate rejection plus live-data safety without network access.
- Verify the replay output plus all five tests.
Replay cached candles
A replay provider implements the same DataProvider boundary as a download provider. It reads existing Candle objects from the cache.
- In data_pipeline.py, find class LiveDataGate:.
- Insert the replay provider directly above that class by copying this code:
class ReplayProvider(DataProvider):
name = "sqlite-replay"
live_approved = False
def __init__(self, cache: CandleCache, source_provider: str):
self.cache = cache
self.source_provider = source_provider
def fetch_history(self, request: DataRequest) -> list[Candle]:
candles = self.cache.load(
self.source_provider,
request.internal_symbol,
request.interval,
)
if not candles:
raise DataUnavailable(
f"No cached candles for {request.internal_symbol} at {request.interval}"
)
return candlesWhat Does This Provider Do?
- The source_provider value selects rows originally stored by the configured provider.
- The cache returns normalized candles using its ascending timestamp query.
- An empty result raises DataUnavailable so downstream code never receives a silent empty replay.
- The provider remains blocked from live decisions because live_approved is false.
- Save data_pipeline.py.
- Confirm that the existing live gate still loads by running this command:
python data_pipeline.py live-checkWhat Does This Check Prove?
The command imports the updated module before running the existing gate. You should still see LIVE BLOCKED: Provider yfinance is not approved for live trading.
Seeing a Python Error?
Check that ReplayProvider sits after CandleCache. Its type annotations depend on that class.
Ask for help with diagnosing the replay provider error.
The command handler builds a request from the watchlist. It initializes the same cache before loading its provider rows.
- In data_pipeline.py, find def run_live_check(settings: Settings) -> int:.
- Insert the replay handler directly above that function by copying this code:
def run_replay(settings: Settings, internal_symbol: str) -> int:
request = build_request(settings, internal_symbol)
cache = CandleCache(settings.cache_path)
cache.initialize()
replay = ReplayProvider(cache, source_provider=settings.provider)
candles = replay.fetch_history(request)
print(f"REPLAYED={len(candles)}")
print(f"FIRST_TIMESTAMP={candles[0].timestamp}")
print(f"LAST_TIMESTAMP={candles[-1].timestamp}")
return 0What Does This Handler Do?
- The request resolves SBIN through the configured watchlist.
- The cache initialization ensures the database exists before replay begins.
- The first timestamp plus last timestamp make the replay order visible.
- Save data_pipeline.py.
- Check that the module still loads by rerunning the existing live check:
python data_pipeline.py live-checkWhat Should You See?
You should see the same controlled live block. This proves the new handler did not weaken the earlier safety path.
The Module Does Not Load?
Check the indentation inside run_replay(). Confirm that the function sits outside every class.
Ask for help with fixing the replay handler.
The parser needs a replay subcommand before the terminal can reach the new handler. The main dispatcher sends the selected symbol to run_replay().
- In build_parser(), locate this existing line:
subparsers.add_parser("live-check")Where Does Replay Belong?
This line marks the end of the existing subcommand declarations. The replay parser belongs immediately above it.
- Replace that line with this parser block:
replay_parser = subparsers.add_parser("replay")
replay_parser.add_argument("--symbol", default="SBIN")
subparsers.add_parser("live-check")What Does This Parser Add?
The new subcommand accepts the same internal symbol style as fetch. Its default selects SBIN when no symbol is supplied.
- In main(), locate this existing branch:
if args.command == "live-check":
return run_live_check(settings)Where Does the Route Belong?
This branch is the final existing command route. The replay route belongs directly above it.
- Replace that branch with this dispatcher block:
if args.command == "replay":
return run_replay(settings, args.symbol)
if args.command == "live-check":
return run_live_check(settings)What Does This Route Do?
The dispatcher passes the parsed symbol into run_replay(). The existing live-check route remains unchanged.
- Save data_pipeline.py.
Before you run the replay, do you expect the first timestamp or the last timestamp to be earlier?
- Replay the cached SBIN candles by running this command:
python data_pipeline.py replay --symbol SBINWhat Should Replay Print?
You should see a positive REPLAYED count. The FIRST_TIMESTAMP value should be earlier than or equal to LAST_TIMESTAMP.
That order comes from the cache query. Every later consumer receives the same timeline.
No Cached Candles to Replay?
Confirm that the earlier fetch populated data/candles.sqlite3. Check that config.toml still maps SBIN to SBIN.NS.
Ask for help with diagnosing an empty replay.
Build offline safety tests
The test suite uses temporary databases plus fixed timestamps. A dummy approved provider tests feed freshness without contacting Yahoo.
- Select the workspace folder in the VS Code Explorer sidebar.
- Click the New Folder icon.
- Enter tests as the folder name.
- Select the new tests folder.
- Click the New File icon.
- Enter __init__.py as the file name.
You should see an empty tests/__init__.py file in the Explorer. The tests directory is now an importable Python package.
- Select the tests folder in the Explorer.
- Click the New File icon.
- Enter test_data_pipeline.py as the file name.
- Add the test imports by copying this code:
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from data_pipeline import (
Candle,
CandleCache,
DataProvider,
DataRequest,
LiveDataBlocked,
LiveDataGate,
ReplayProvider,
YFinanceProvider,
)What Do These Imports Provide?
The standard-library imports provide the test framework plus isolated temporary directories. The project imports expose the cache, replay provider, plus live gate under test.
- Save tests/test_data_pipeline.py.
- Confirm that test discovery can import the project by running this command:
python -m unittest -vWhat Does This First Run Prove?
The run should finish without an import failure. The suite has no test methods yet.
Test Discovery Cannot Import the Pipeline?
Confirm that tests sits beside data_pipeline.py. Check every imported identifier against the names in the pipeline.
Ask for help with fixing the test imports.
The dummy provider represents a feed with explicit live permission. Fixed candles give the cache tests predictable values.
- Add the approved dummy provider below the import block by copying this code:
class DummyLiveProvider(DataProvider):
name = "dummy-live"
live_approved = True
def fetch_history(self, request: DataRequest) -> list[Candle]:
return []Why Is This Provider Approved?
The test double isolates feed freshness from vendor permission. Only this dummy provider receives explicit approval.
- Add the candle factory below DummyLiveProvider by copying this code:
def make_candle(timestamp: str, close: float = 100.0) -> Candle:
return Candle(
provider="yfinance",
internal_symbol="SBIN",
provider_symbol="SBIN.NS",
interval="5m",
timestamp=timestamp,
open=close - 1.0,
high=close + 1.0,
low=close - 2.0,
close=close,
volume=1000,
)What Does the Factory Control?
Each generated candle has valid OHLC relationships plus a fixed volume. Tests can vary the timestamp or closing price without repeating the full object.
- Save tests/test_data_pipeline.py.
- Confirm that both helpers import cleanly by rerunning the suite:
python -m unittest -vWhat Does This Run Prove?
The test module should load without errors. The helpers are ready for the five test methods.
The Helpers Do Not Load?
Check the indentation inside DummyLiveProvider. Confirm that make_candle() begins at the left edge of the file.
Ask for help with correcting the test helpers.
The first test writes the same candle twice. The composite primary key plus INSERT OR IGNORE must leave one stored row.
- Add the test class plus duplicate-cache test below make_candle() by copying this code:
class DataPipelineTests(unittest.TestCase):
def test_cache_rejects_duplicate_candles(self) -> None:
with TemporaryDirectory() as directory:
cache = CandleCache(Path(directory) / "candles.sqlite3")
cache.initialize()
candle = make_candle("2026-10-27T09:15:00+05:30")
self.assertEqual(cache.store([candle]), 1)
self.assertEqual(cache.store([candle]), 0)
self.assertEqual(cache.count("yfinance", "SBIN", "5m"), 1)What Does This Test Protect?
The first store must insert one row. The retry must insert zero rows while the final count remains one.
- Save tests/test_data_pipeline.py.
- Run the suite to execute the duplicate-cache test:
python -m unittest -vWhat Should Pass?
The output should list test_cache_rejects_duplicate_candles as passing. The run should finish with OK.
The Duplicate Test Failed?
Confirm that CandleCache.store() uses INSERT OR IGNORE. Check that the primary key includes provider, internal symbol, interval, plus timestamp.
Ask for help with diagnosing duplicate cache rows.
Replay order must depend on the cache query. The next test inserts candles out of order so insertion-order replay becomes visible.
- Inside DataPipelineTests, add the replay-order test below the duplicate test by copying this code:
def test_replay_is_sorted_by_timestamp(self) -> None:
with TemporaryDirectory() as directory:
cache = CandleCache(Path(directory) / "candles.sqlite3")
cache.initialize()
later = make_candle("2026-10-27T09:20:00+05:30", 101.0)
earlier = make_candle("2026-10-27T09:15:00+05:30", 100.0)
cache.store([later, earlier])
replay = ReplayProvider(cache, source_provider="yfinance")
request = DataRequest("SBIN", "SBIN.NS", "5m", "5d")
candles = replay.fetch_history(request)
self.assertEqual([candle.timestamp for candle in candles], [
"2026-10-27T09:15:00+05:30",
"2026-10-27T09:20:00+05:30",
])What Does This Test Protect?
The test stores the later candle first. Replay must still return the earlier timestamp before the later timestamp.
- Save tests/test_data_pipeline.py.
- Run the suite to test deterministic ordering:
python -m unittest -vWhat Should Pass Now?
The output should list two passing tests. The replay test proves that timestamp sorting happens before a strategy receives the candles.
The Replay Order Is Reversed?
Check that CandleCache.load() ends its query with ORDER BY timestamp ASC.
Ask for help with fixing deterministic replay.
A successful research download must never grant live permission. The third test locks that provider capability into the suite.
- Inside DataPipelineTests, add the yfinance blocking test below the replay test by copying this code:
def test_yfinance_is_blocked_for_live_mode(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
with self.assertRaises(LiveDataBlocked):
gate.require_ready(YFinanceProvider(), "live", None, now_monotonic=100.0)What Does This Test Protect?
The test passes only when YFinanceProvider raises LiveDataBlocked in live mode. Download success cannot change that permission.
- Save tests/test_data_pipeline.py.
- Run the suite to check the research-provider block:
python -m unittest -vWhat Should Pass Now?
The output should list three passing tests. The yfinance check remains offline because it never calls fetch_history().
Yfinance Passed the Live Gate?
Confirm that YFinanceProvider.live_approved remains false. Check that require_ready() tests provider approval before feed age.
Ask for help with restoring the fail-closed provider check.
Provider approval covers identity. Feed freshness covers whether approved data is timely enough for live decisions.
- Inside DataPipelineTests, add the stale-feed test below the yfinance test by copying this code:
def test_stale_live_feed_is_blocked(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
with self.assertRaises(LiveDataBlocked):
gate.require_ready(
DummyLiveProvider(),
"live",
last_message_monotonic=94.0,
now_monotonic=100.0,
)What Does This Test Protect?
The fixed monotonic values create a message age of six seconds. That exceeds the five-second boundary.
The approved provider must still be blocked. Permission never overrides freshness.
- Save tests/test_data_pipeline.py.
- Run the suite to exercise stale-feed blocking:
python -m unittest -vWhat Should Pass Now?
The output should list four passing tests. The stale-feed test proves that approval alone cannot bypass the age threshold.
The Stale Feed Was Approved?
Confirm that the gate blocks ages greater than stale_after_seconds. Check that the test uses 94.0 plus 100.0 for its monotonic values.
Ask for help with diagnosing the stale-feed boundary.
A fail-closed gate also needs a passing path for healthy approved data. The final test places a message four seconds behind the fixed current time.
- Inside DataPipelineTests, add the fresh-feed test plus test runner below the stale-feed test by copying this code:
def test_fresh_live_feed_is_approved(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
gate.require_ready(
DummyLiveProvider(),
"live",
last_message_monotonic=96.0,
now_monotonic=100.0,
)
if __name__ == "__main__":
unittest.main()What Does the Final Test Protect?
The four-second message age stays within the configured boundary. The test passes only when the approved feed is accepted.
The entry point also lets the test module run directly. Standard discovery continues through the same test class.
- Save tests/test_data_pipeline.py.
- Run all five offline tests:
python -m unittest -vWhat Should the Full Suite Show?
The terminal should list five passing tests. The final line should show OK without any network download.
One of the Five Tests Failed?
Use the failing test name to identify the broken rule. Compare that method with its matching cache, replay, or live-gate implementation.
Ask for help with diagnosing the failed safety test.
Verify the final workspace
The final workspace records the two local cleanup commands. This keeps the generated database plus virtual environment easy to remove after the project.
- In README.md, add this cleanup section after the run commands:
## Cleanup
```bash
deactivate
rm -rf .venv data/candles.sqlite3
```What Does This Section Record?
The first command exits the active environment. The second command removes only the local environment plus generated candle database.
- Save README.md.
- Compare the final project files with the complete reference below.
✔️ Awesome, I've got everything!
Great. Confirm that every file is saved before the final replay plus test run.
ⓧ I'd like to double check the full code
pandas==3.1.0
yfinance==1.7.0mode = "research"
provider = "yfinance"
interval = "5m"
period = "5d"
cache_path = "data/candles.sqlite3"
stale_after_seconds = 5
[risk]
max_daily_loss_rupees = 50
[watchlist]
HDFCBANK = "HDFCBANK.NS"
ICICIBANK = "ICICIBANK.NS"
SBIN = "SBIN.NS"
RELIANCE = "RELIANCE.NS"
INFY = "INFY.NS"
ITC = "ITC.NS"
TATASTEEL = "TATASTEEL.NS"from __future__ import annotations
import argparse
import math
import sqlite3
import time
import tomllib
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import yfinance as yf
class DataPipelineError(RuntimeError):
"""Base error for controlled data-pipeline failures."""
class DataUnavailable(DataPipelineError):
"""Raised when a provider cannot supply valid candles."""
class LiveDataBlocked(DataPipelineError):
"""Raised when a provider or feed is unsafe for live decisions."""
@dataclass(frozen=True)
class DataRequest:
internal_symbol: str
provider_symbol: str
interval: str
period: str
@dataclass(frozen=True)
class Candle:
provider: str
internal_symbol: str
provider_symbol: str
interval: str
timestamp: str
open: float
high: float
low: float
close: float
volume: int
@dataclass(frozen=True)
class Settings:
mode: str
provider: str
interval: str
period: str
cache_path: Path
stale_after_seconds: float
max_daily_loss_rupees: int
watchlist: dict[str, str]
def load_settings(path: Path) -> Settings:
with path.open("rb") as config_file:
raw = tomllib.load(config_file)
mode = str(raw["mode"]).lower()
if mode not in {"research", "paper", "live"}:
raise DataPipelineError(f"Unsupported mode: {mode}")
stale_after_seconds = float(raw["stale_after_seconds"])
if stale_after_seconds <= 0:
raise DataPipelineError("stale_after_seconds must be positive")
max_daily_loss_rupees = int(raw["risk"]["max_daily_loss_rupees"])
if max_daily_loss_rupees <= 0:
raise DataPipelineError("max_daily_loss_rupees must be positive")
watchlist = {
str(symbol).upper(): str(provider_symbol)
for symbol, provider_symbol in raw["watchlist"].items()
}
if not watchlist:
raise DataPipelineError("watchlist must not be empty")
return Settings(
mode=mode,
provider=str(raw["provider"]).lower(),
interval=str(raw["interval"]),
period=str(raw["period"]),
cache_path=Path(str(raw["cache_path"])),
stale_after_seconds=stale_after_seconds,
max_daily_loss_rupees=max_daily_loss_rupees,
watchlist=watchlist,
)
class DataProvider(ABC):
name = "unconfigured"
live_approved = False
@abstractmethod
def fetch_history(self, request: DataRequest) -> list[Candle]:
"""Return normalized candles or raise a controlled error."""
class YFinanceProvider(DataProvider):
name = "yfinance"
live_approved = False
def fetch_history(self, request: DataRequest) -> list[Candle]:
frame = yf.download(
tickers=request.provider_symbol,
period=request.period,
interval=request.interval,
actions=False,
threads=False,
auto_adjust=False,
progress=False,
timeout=10,
multi_level_index=False,
)
if frame is None or frame.empty:
raise DataUnavailable(
f"No candles returned for {request.provider_symbol}. "
"The symbol, Yahoo availability, network, or rate limit may be the cause."
)
required_columns = ["Open", "High", "Low", "Close", "Volume"]
missing = [column for column in required_columns if column not in frame.columns]
if missing:
raise DataUnavailable(f"Missing required columns: {missing}")
candles: list[Candle] = []
selected = frame[required_columns]
for row in selected.itertuples(index=True, name=None):
timestamp, open_price, high, low, close, volume = row
numeric_values = [
float(open_price),
float(high),
float(low),
float(close),
float(volume),
]
if not all(math.isfinite(value) for value in numeric_values):
raise DataUnavailable(
f"Non-finite candle received for {request.provider_symbol} at {timestamp}"
)
candle = Candle(
provider=self.name,
internal_symbol=request.internal_symbol,
provider_symbol=request.provider_symbol,
interval=request.interval,
timestamp=timestamp.isoformat(),
open=numeric_values[0],
high=numeric_values[1],
low=numeric_values[2],
close=numeric_values[3],
volume=int(numeric_values[4]),
)
validate_candle(candle)
candles.append(candle)
if not candles:
raise DataUnavailable(f"No valid candles remained for {request.provider_symbol}")
return candles
def validate_candle(candle: Candle) -> None:
prices = [candle.open, candle.high, candle.low, candle.close]
if any(price <= 0 for price in prices):
raise DataUnavailable(f"Non-positive price at {candle.timestamp}")
if candle.volume < 0:
raise DataUnavailable(f"Negative volume at {candle.timestamp}")
if candle.low > candle.high:
raise DataUnavailable(f"Low exceeds high at {candle.timestamp}")
if not candle.low <= candle.open <= candle.high:
raise DataUnavailable(f"Open falls outside low/high at {candle.timestamp}")
if not candle.low <= candle.close <= candle.high:
raise DataUnavailable(f"Close falls outside low/high at {candle.timestamp}")
class CandleCache:
def __init__(self, path: Path):
self.path = path
def initialize(self) -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
connection = sqlite3.connect(self.path)
try:
connection.execute(
"""
CREATE TABLE IF NOT EXISTS candles (
provider TEXT NOT NULL,
internal_symbol TEXT NOT NULL,
provider_symbol TEXT NOT NULL,
interval TEXT NOT NULL,
timestamp TEXT NOT NULL,
open REAL NOT NULL,
high REAL NOT NULL,
low REAL NOT NULL,
close REAL NOT NULL,
volume INTEGER NOT NULL,
downloaded_at TEXT NOT NULL,
PRIMARY KEY (provider, internal_symbol, interval, timestamp)
)
"""
)
connection.commit()
finally:
connection.close()
def store(self, candles: list[Candle]) -> int:
if not candles:
return 0
downloaded_at = datetime.now(timezone.utc).isoformat()
rows = [
(
candle.provider,
candle.internal_symbol,
candle.provider_symbol,
candle.interval,
candle.timestamp,
candle.open,
candle.high,
candle.low,
candle.close,
candle.volume,
downloaded_at,
)
for candle in candles
]
connection = sqlite3.connect(self.path)
try:
before = connection.total_changes
connection.executemany(
"""
INSERT OR IGNORE INTO candles (
provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume, downloaded_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
rows,
)
connection.commit()
return connection.total_changes - before
finally:
connection.close()
def count(self, provider: str, internal_symbol: str, interval: str) -> int:
connection = sqlite3.connect(self.path)
try:
result = connection.execute(
"""
SELECT COUNT(*)
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
""",
(provider, internal_symbol, interval),
).fetchone()
return int(result[0])
finally:
connection.close()
def load(self, provider: str, internal_symbol: str, interval: str) -> list[Candle]:
connection = sqlite3.connect(self.path)
try:
rows = connection.execute(
"""
SELECT provider, internal_symbol, provider_symbol, interval,
timestamp, open, high, low, close, volume
FROM candles
WHERE provider = ? AND internal_symbol = ? AND interval = ?
ORDER BY timestamp ASC
""",
(provider, internal_symbol, interval),
).fetchall()
finally:
connection.close()
return [Candle(*row) for row in rows]
class ReplayProvider(DataProvider):
name = "sqlite-replay"
live_approved = False
def __init__(self, cache: CandleCache, source_provider: str):
self.cache = cache
self.source_provider = source_provider
def fetch_history(self, request: DataRequest) -> list[Candle]:
candles = self.cache.load(
self.source_provider,
request.internal_symbol,
request.interval,
)
if not candles:
raise DataUnavailable(
f"No cached candles for {request.internal_symbol} at {request.interval}"
)
return candles
class LiveDataGate:
def __init__(self, stale_after_seconds: float):
if stale_after_seconds <= 0:
raise ValueError("stale_after_seconds must be positive")
self.stale_after_seconds = stale_after_seconds
def require_ready(
self,
provider: DataProvider,
mode: str,
last_message_monotonic: float | None,
now_monotonic: float | None = None,
) -> None:
if mode != "live":
return
if not provider.live_approved:
raise LiveDataBlocked(
f"Provider {provider.name} is not approved for live trading"
)
if last_message_monotonic is None:
raise LiveDataBlocked("No live-feed message has been received")
now = time.monotonic() if now_monotonic is None else now_monotonic
age = now - last_message_monotonic
if age < 0:
raise LiveDataBlocked("Monotonic feed clock moved backwards")
if age > self.stale_after_seconds:
raise LiveDataBlocked(
f"Feed age {age:.1f}s exceeds {self.stale_after_seconds:.1f}s"
)
def build_request(settings: Settings, internal_symbol: str) -> DataRequest:
symbol = internal_symbol.upper()
if symbol not in settings.watchlist:
raise DataPipelineError(f"Symbol {symbol} is outside the configured watchlist")
return DataRequest(
internal_symbol=symbol,
provider_symbol=settings.watchlist[symbol],
interval=settings.interval,
period=settings.period,
)
def run_fetch(settings: Settings, internal_symbol: str) -> int:
if settings.provider != "yfinance":
raise DataPipelineError(f"Unsupported provider: {settings.provider}")
request = build_request(settings, internal_symbol)
provider = YFinanceProvider()
candles = provider.fetch_history(request)
cache = CandleCache(settings.cache_path)
cache.initialize()
first_inserted = cache.store(candles)
duplicate_inserted = cache.store(candles)
cached_rows = cache.count(provider.name, request.internal_symbol, request.interval)
print(f"FETCHED={len(candles)}")
print(f"FIRST_INSERTED={first_inserted}")
print(f"DUPLICATE_INSERTED={duplicate_inserted}")
print(f"CACHE_ROWS={cached_rows}")
print(f"LAST_CANDLE={candles[-1]}")
return 0
def run_replay(settings: Settings, internal_symbol: str) -> int:
request = build_request(settings, internal_symbol)
cache = CandleCache(settings.cache_path)
cache.initialize()
replay = ReplayProvider(cache, source_provider=settings.provider)
candles = replay.fetch_history(request)
print(f"REPLAYED={len(candles)}")
print(f"FIRST_TIMESTAMP={candles[0].timestamp}")
print(f"LAST_TIMESTAMP={candles[-1].timestamp}")
return 0
def run_live_check(settings: Settings) -> int:
gate = LiveDataGate(settings.stale_after_seconds)
provider = YFinanceProvider()
try:
gate.require_ready(
provider=provider,
mode="live",
last_message_monotonic=None,
)
except LiveDataBlocked as error:
print(f"LIVE BLOCKED: {error}")
return 0
raise DataPipelineError("Unsafe result: yfinance unexpectedly passed the live gate")
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(description="Fail-closed NSE data pipeline")
parser.add_argument("--config", default="config.toml")
subparsers = parser.add_subparsers(dest="command", required=True)
fetch_parser = subparsers.add_parser("fetch")
fetch_parser.add_argument("--symbol", default="SBIN")
replay_parser = subparsers.add_parser("replay")
replay_parser.add_argument("--symbol", default="SBIN")
subparsers.add_parser("live-check")
return parser
def main() -> int:
args = build_parser().parse_args()
settings = load_settings(Path(args.config))
print(f"MODE={settings.mode}")
print(f"FUTURE_MAX_DAILY_LOSS_RUPEES={settings.max_daily_loss_rupees}")
try:
if args.command == "fetch":
return run_fetch(settings, args.symbol)
if args.command == "replay":
return run_replay(settings, args.symbol)
if args.command == "live-check":
return run_live_check(settings)
except DataPipelineError as error:
print(f"DATA PIPELINE HALTED: {error}")
return 1
raise DataPipelineError(f"Unknown command: {args.command}")
if __name__ == "__main__":
raise SystemExit(main())The tests/__init__.py file should be empty.
import unittest
from pathlib import Path
from tempfile import TemporaryDirectory
from data_pipeline import (
Candle,
CandleCache,
DataProvider,
DataRequest,
LiveDataBlocked,
LiveDataGate,
ReplayProvider,
YFinanceProvider,
)
class DummyLiveProvider(DataProvider):
name = "dummy-live"
live_approved = True
def fetch_history(self, request: DataRequest) -> list[Candle]:
return []
def make_candle(timestamp: str, close: float = 100.0) -> Candle:
return Candle(
provider="yfinance",
internal_symbol="SBIN",
provider_symbol="SBIN.NS",
interval="5m",
timestamp=timestamp,
open=close - 1.0,
high=close + 1.0,
low=close - 2.0,
close=close,
volume=1000,
)
class DataPipelineTests(unittest.TestCase):
def test_cache_rejects_duplicate_candles(self) -> None:
with TemporaryDirectory() as directory:
cache = CandleCache(Path(directory) / "candles.sqlite3")
cache.initialize()
candle = make_candle("2026-10-27T09:15:00+05:30")
self.assertEqual(cache.store([candle]), 1)
self.assertEqual(cache.store([candle]), 0)
self.assertEqual(cache.count("yfinance", "SBIN", "5m"), 1)
def test_replay_is_sorted_by_timestamp(self) -> None:
with TemporaryDirectory() as directory:
cache = CandleCache(Path(directory) / "candles.sqlite3")
cache.initialize()
later = make_candle("2026-10-27T09:20:00+05:30", 101.0)
earlier = make_candle("2026-10-27T09:15:00+05:30", 100.0)
cache.store([later, earlier])
replay = ReplayProvider(cache, source_provider="yfinance")
request = DataRequest("SBIN", "SBIN.NS", "5m", "5d")
candles = replay.fetch_history(request)
self.assertEqual([candle.timestamp for candle in candles], [
"2026-10-27T09:15:00+05:30",
"2026-10-27T09:20:00+05:30",
])
def test_yfinance_is_blocked_for_live_mode(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
with self.assertRaises(LiveDataBlocked):
gate.require_ready(YFinanceProvider(), "live", None, now_monotonic=100.0)
def test_stale_live_feed_is_blocked(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
with self.assertRaises(LiveDataBlocked):
gate.require_ready(
DummyLiveProvider(),
"live",
last_message_monotonic=94.0,
now_monotonic=100.0,
)
def test_fresh_live_feed_is_approved(self) -> None:
gate = LiveDataGate(stale_after_seconds=5)
gate.require_ready(
DummyLiveProvider(),
"live",
last_message_monotonic=96.0,
now_monotonic=100.0,
)
if __name__ == "__main__":
unittest.main()# Fail-Closed NSE Data Pipeline
This is Stage 1 of a safety-first intraday trading system. It downloads research data, validates it, caches it, replays it, and refuses to approve yfinance for live trading.
It does not contain strategies, a broker adapter, or order placement.
## Linux setup
Verify Visual Studio Code:
```bash
code --version
```
Install uv 0.13.6 and create the Python 3.11 environment:
```bash
curl -LsSf https://astral.sh/uv/0.13.6/install.sh | sh
uv python install 3.11
uv venv --python 3.11
source .venv/bin/activate
uv pip sync requirements.txt
```
## Run the pipeline
```bash
python data_pipeline.py fetch --symbol SBIN
python data_pipeline.py replay --symbol SBIN
python data_pipeline.py live-check
python -m unittest -v
```
## Cleanup
```bash
deactivate
rm -rf .venv data/candles.sqlite3
```Before you run the final check, do you expect replay plus all five tests to follow the same deterministic safety rules?
- Run the replay plus complete offline suite with these commands:
python data_pipeline.py replay --symbol SBIN
python -m unittest -vWhat Does Final Success Look Like?
The replay should print a positive count plus ordered first and last timestamps. The test runner should list five tests before finishing with OK.
The tests use temporary databases plus fixed monotonic times. No test depends on a live network response.
The Final Check Did Not Pass?
If replay fails, confirm that the earlier fetch populated data/candles.sqlite3. If a test fails, use its name to trace the matching safety rule.
Ask for help with diagnosing the final replay or test result.
That completes the safety foundation. Your cache now replays one deterministic timeline while five offline tests defend deduplication plus live-data blocking.
Secret mission
Simulate a Feed Going Stale
Create an approved simulated live provider. Prove that the five-second gate accepts a four-second message. Then prove that it blocks a six-second message.
Clean Up Your Resources
Clean Up Your Resources
Your generated project state stays on your Linux device at a cost of ₹0. Choose whether to keep it available, pause the active environment, or delete the generated local state.
Resources you used:
- A repository-local .venv environment using Python 3.11.
- A local SQLite database at data/candles.sqlite3 containing validated candle rows.
Keep everything running
No action is needed. Choose this if you plan to reuse the pipeline in a later backtester project.
- Keep .venv to preserve the pinned local dependencies.
- Keep data/candles.sqlite3 to preserve validated candles for deterministic replay.
The pipeline remains ready for your next stage. These local files create no ongoing charges.
Pause - I'll come back to this later
Pausing ends the current shell's use of the virtual environment. Your project files remain available for the next session.
- Leave the active virtual environment by running this command:
deactivateWhat Does Pausing Do?
This command exits the virtual environment in your current shell. It leaves .venv on disk.
Your cached candles remain stored at data/candles.sqlite3.
Your shell prompt should stop showing the active environment indicator. You're safely paused with every cached candle still available.
Delete - I don't want to use this again
Deleting these two generated items is permanent. Your final source files, tests, and config.toml stay in place.
- Leave the active virtual environment by running this command:
deactivateWhy Deactivate First?
This command releases the current shell from .venv. The environment directory remains on disk until the next command removes it.
- Remove the virtual environment and candle database by running this command:
rm -rf .venv data/candles.sqlite3What Does Deletion Remove?
The shell removes .venv. It also removes data/candles.sqlite3.
Your pipeline code, configuration, tests, and stale-feed simulation remain unchanged.
- Return to Visual Studio Code from earlier.
- Use the file sidebar to confirm that .venv no longer appears beside data_pipeline.py.
- Expand data in the file sidebar.
- Confirm that data/candles.sqlite3 no longer appears.
That's the local cleanup complete. Your reusable source code remains ready if you decide to rebuild the environment later.
Nice Work!
Nice Work!
You did it! Your fail-closed NSE data pipeline now protects every candle before it can reach future strategy or broker code.
You've learned how to:
- Create a vendor-swappable DataProvider boundary. Apply defensive validation before downstream code can use each normalized five-minute NSE candle.
- Store one canonical candle per key in an idempotent cache built with SQLite. Replay the cached rows in deterministic timestamp order.
- Keep yfinance outside the future live-order path with a fail-closed gate. Prove the safety rules with five offline tests.
- Secret Mission: Approve SimulatedLiveProvider when its message is four seconds old. Block the same feed when its message is six seconds old.
Ready to quiz yourself? Test how well the pipeline's safety rules stuck.