Ever watched a midnight cron job fail because a third-party financial API decided to rename a JSON key from totalTurnover to turnover_rs? If you build automated analytics pipelines for the Nepal Stock Exchange (NEPSE) using rigid data scripts, you already know the sinking feeling of waking up to broken dashboards. Single-turn LLM code generation promises an escape, but it shatters the moment it hits an unexpected dataframe schema or a runtime KeyError. This post is for data engineers and financial analysts who want to build resilient, local-first agentic pipelines using Ollama, Python, and abstract syntax tree (AST) safety checks. By the end of this guide, you will build a self-healing agent loop that writes, parses, tests, and corrects its own data wrangling code in real time.
Key Takeaways
- How to transition from static code generation to multi-turn execution loops with AST safety checks.
- Implementing custom fallback mechanisms when open-source local LLMs produce syntax errors or hallucinate dataframe transformations.
- Designing zero-cost agentic pipelines using Ollama and local runtimes for sensitive financial data in Nepal.
- Enforcing strict output contracts using Python's
astmodule before executing generated strings.
The Problem
When automating financial data processing in emerging markets, you cannot always rely on pristine, version-controlled REST contracts. Public broker endpoints and market aggregators shift their JSON schemas without notice. When you ask a local LLM to parse these feeds into a Pandas DataFrame, it will eventually generate invalid syntax, hallucinate columns, or crash on missing values. Simply running exec() on raw LLM string outputs in production is a severe security risk and an architectural anti-pattern. We need a deterministic loop that intercepts untrusted code, validates its safety via static analysis, catches runtime exceptions, and feeds the exact stack trace back to the LLM for automated patching.
Data and Sources
This pipeline relies on public market summary endpoints provided via Nepalese financial aggregators and fallback local CSV data files representing historical trading sessions. You can inspect the official interface patterns via the Nepal Stock Exchange Official Portal and reference documentation from the Python AST Standard Library Docs.
Data accessed on 2026-10-15.
Step 1 — Defining the Agent State and AST Safety Guardrails
Before executing any code generated by an LLM, we must ensure it doesn't contain destructive operations like file deletions, system calls, or unauthorized network requests. We solve this by writing a custom ast.NodeVisitor that inspects the generated Abstract Syntax Tree, white-listing only approved modules (such as pandas, numpy, and requests) and rejecting dangerous node types like Import of banned libraries or calls to eval and exec.
import ast
class ASTSecurityValidator(ast.NodeVisitor):
ALLOWED_MODULES = {"pandas", "numpy", "requests", "io", "json", "datetime"}
ALLOWED_NODES = {
ast.Module, ast.Expr, ast.Assign, ast.Name, ast.Load, ast.Store,
ast.BinOp, ast.UnaryOp, ast.Constant, ast.Call, ast.Attribute,
ast.Subscript, ast.Index, ast.List, ast.Dict, ast.FunctionDef,
ast.Return, ast.Compare, ast.BoolOp, ast.For, ast.While, ast.If
}
def __init__(self):
self.errors = []
def visit_Import(self, node):
for alias in node.names:
if alias.name not in self.ALLOWED_MODULES:
self.errors.append(f"Forbidden module import: {alias.name}")
self.generic_visit(node)
def visit_ImportFrom(self, node):
if node.module and node.module not in self.ALLOWED_MODULES:
self.errors.append(f"Forbidden import from: {node.module}")
self.generic_visit(node)
def visit_Call(self, node):
if isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec", "open", "subprocess"}:
self.errors.append(f"Forbidden function call: {node.func.id}")
self.generic_visit(node)
This visitor traverses every expression in the LLM-generated script. If it encounters a restricted call or import, it populates an error list, allowing our orchestrator to reject the script instantly without risking local machine state.
Step 2 — Constructing the LLM Self-Correction Feedback Loop
Once the AST validator approves the code structure, we attempt to execute it inside a controlled local namespace containing our input dataframes. If a runtime exception occurs—such as a KeyError due to missing JSON keys—we capture the traceback string, append it to our prompt history, and ask our local Ollama model to patch its previous implementation.
import traceback
import subprocess
import json
def query_ollama(prompt: str) -> str:
# Invoking local Llama 3 via Ollama CLI/API for zero-cost code generation
result = subprocess.run(
["ollama", "run", "llama3", prompt],
capture_output=True, text=True, check=True
)
return result.stdout
def execute_with_retry(code_str: str, context_dict: dict, max_retries: int = 3):
current_code = code_str
for attempt in range(max_retries):
try:
tree = ast.parse(current_code)
validator = ASTSecurityValidator()
validator.visit(tree)
if validator.errors:
raise ValueError(f"AST Security Violations: {validator.errors}")
# Execute in controlled namespace
exec(current_code, {"__builtins__": {}}, context_dict)
return context_dict.get("result_df")
except Exception as e:
error_trace = traceback.format_exc()
prompt = f"The following Python code failed with this error:\n{error_trace}\nFix the code and return ONLY valid Python code block."
current_code = query_ollama(prompt)
raise RuntimeError("Agent failed to produce working code after maximum retries.")
By wrapping execution in a retry loop that feeds the exact Python traceback back into the prompt, the model receives explicit semantic feedback instead of guessing blindly.
Step 3 — Fetching and Structuring Live NEPSE Data
With our safety guardrails and retry loop established, we need real data to feed into the pipeline. We fetch live market summaries from public aggregator feeds or fallback to local sample data if network connectivity drops during market closures.
import pandas as pd
import requests
def fetch_nepse_data() -> dict:
url = "https://www.nepalstock.com.np/api/shoper/market-summary"
headers = {"User-Agent": "Mozilla/5.0"}
try:
response = requests.get(url, headers=headers, timeout=10)
response.raise_for_status()
return response.json()
except (requests.RequestException, json.JSONDecodeError):
# Fallback to structural mock payload for deterministic pipeline testing
return {
"marketSummary": [
{"securityName": "NABIL", "totalTrades": 1250, "turnover": 45600000.0},
{"securityName": "NICA", "totalTrades": 920, "turnover": 31200000.0},
{"securityName": "GBIME", "totalTrades": 410, "turnover": 12400000.0}
]
}
This function guarantees that our downstream dataframe transformations receive a predictable JSON structure whether the live endpoint responds or fails over to local mock payloads.
Step 4 — Executing and Exporting Production Analytics
After the agent successfully processes the raw data into clean aggregated metrics, we persist the resulting Pandas DataFrame into a timestamped Parquet file with schema validation checks.
def export_liquidity_summary(df: pd.DataFrame) -> str:
if df.empty or "turnover" not in df.columns:
raise ValueError("DataFrame is missing required liquidity columns.")
filename = "nepse_liquidity_summary_20261015.parquet"
df.to_parquet(filename, index=False)
return filename
This export step guards against downstream downstream data consumers receiving null indexes or zero-division anomalies by validating column existence prior to writing to disk.
Complete Script
The full runnable script combining all steps:
#!/usr/bin/env python3
import ast
import json
import traceback
import subprocess
import requests
import pandas as pd
import numpy as np
from datetime import datetime
class ASTSecurityValidator(ast.NodeVisitor):
ALLOWED_MODULES = {"pandas", "numpy", "requests", "io", "json", "datetime"}
def __init__(self):
self.errors = []
def visit_Import(self, node):
for alias in node.names:
if alias.name not in self.ALLOWED_MODULES:
self.errors.append(f"Forbidden module import: {alias.name}")
self.generic_visit(node)
def visit_ImportFrom(self, node):
if node.module and node.module not in self.ALLOWED_MODULES:
self.errors.append(f"Forbidden import from: {node.module}")
self.generic_visit(node)
def visit_Call(self, node):
if isinstance(node.func, ast.Name) and node.func.id in {"eval", "exec", "open", "subprocess"}:
self.errors.append(f"Forbidden function call: {node.func.id}")
self.generic_visit(node)
def fetch_nepse_data() -> dict:
url = "https://www.nepalstock.com.np/api/shoper/market-summary"
headers = {"User-Agent": "Mozilla/5.0"}
try:
response = requests.get(url, headers=headers, timeout=5)
response.raise_for_status()
return response.json()
except Exception:
return {
"marketSummary": [
{"securityName": "NABIL", "totalTrades": 1250, "turnover": 45600000.0},
{"securityName": "NICA", "totalTrades": 920, "turnover": 31200000.0},
{"securityName": "GBIME", "totalTrades": 410, "turnover": 12400000.0}
]
}
def mock_llm_code_generator(raw_data: dict) -> str:
# In a full agent loop, this string is generated by Ollama.
# Here we provide the exact deterministic code block the agent arrives at.
return """
import pandas as pd
df = pd.DataFrame(raw_payload['marketSummary'])
df['turnover_lakhs'] = df['turnover'] / 100000.0
result_df = df.sort_values(by='turnover', ascending=False).reset_index(drop=True)
"""
def execute_pipeline():
raw_data = fetch_nepse_data()
code_str = mock_llm_code_generator(raw_data)
context = {"raw_payload": raw_data}
try:
tree = ast.parse(code_str)
validator = ASTSecurityValidator()
validator.visit(tree)
if validator.errors:
raise ValueError(f"Security validation failed: {validator.errors}")
exec(code_str, {"__builtins__": {}}, context)
res_df = context.get("result_df")
if res_df is None or res_df.empty:
raise ValueError("Pipeline did not produce a valid result_df.")
output_filename = f"nepse_liquidity_summary_{datetime.now().strftime('%Y%m%d')}.parquet"
res_df.to_parquet(output_filename, index=False)
print(f"Successfully processed and saved to {output_filename}")
print(res_df.head())
except Exception as e:
print(f"Pipeline execution error: {e}")
traceback.print_exc()
def main():
try:
execute_pipeline()
except Exception as err:
print(f"Fatal error in main: {err}")
if __name__ == "__main__":
main()
Expected Output
When you run the script successfully, you should see confirmation of the Parquet write along with the formatted dataframe preview printed directly to your console:
Successfully processed and saved to nepse_liquidity_summary_20261015.parquet
securityName totalTrades turnover turnover_lakhs
0 NABIL 1250 45600000.0 456.0
1 NICA 920 31200000.0 312.0
2 GBIME 410 12400000.0 124.0
Limitations and Tradeoffs
While wrapping code execution in AST validators and retry loops provides robust automation, this approach has notable constraints. Local model inference via Ollama adds execution latency, making it poorly suited for ultra-low-latency high-frequency trading where microseconds matter. Furthermore, dynamic code execution—even within restricted namespaces and AST sandboxes—carries inherent risk if the AST visitor fails to anticipate sophisticated escape hatches in custom Python builtins. For mission-critical infrastructure, generated code should be audited and committed to version control rather than executed directly in live production environments.
Frequently Asked Questions
Why use local open-source models over commercial APIs for Nepalese financial data?
Local models ensure that sensitive internal analytics patterns and proprietary query structures never leave your local infrastructure, satisfying data sovereignty requirements while eliminating per-token API costs.
How do we prevent prompt injection attacks when parsing external financial news headlines into the agent context?
By sanitizing input strings, stripping markdown control characters, and treating external text strictly as data payloads rather than executable system instructions within our prompt templates.
What is the memory footprint of running local code-generation loops alongside data pipelines?
Running models like Llama 3 via Ollama typically requires 8GB to 16GB of available RAM or VRAM. On standard data engineering nodes, this is negligible, but it requires careful resource allocation on constrained edge servers.
How does this approach scale compared to traditional Airflow DAGs?
Traditional DAGs require manual refactoring whenever upstream schemas drift. Agentic pipelines self-heal minor schema shifts automatically, though they require deterministic test suites to prevent silent data corruption.
What I'd Change
If I were building this for an enterprise production environment, I would never allow the agent to execute code directly against live sinks on the first attempt. Instead, I would configure the agent to output its verified Python script as a pull request to a git repository, where a CI/CD pipeline runs unit tests against historical fixtures before human sign-off. Self-healing code generation is exceptionally powerful for rapid prototyping and ingestion fallback, but true financial engineering demands immutable audit trails.