Building a quant research pipeline with AI agents

df['signal'] = np.where(df['close'].shift(–1) > df['close'], 1, –1)

# The agent forgot to group by symbol before calculating rolling z-scores
df['z_score'] = (df['close'] – df['close'].rolling(20).mean()) / df['close'].rolling(20).std()

import os
import sys
import ast
import json
import traceback
import numpy as np
import pandas as pd
from typing import Dict, Any, Tuple
import openai

# Secure your API key before running
# os.environ["OPENAI_API_KEY"] = "your-actual-api-key"

# =====================================================================
# 1. AST Linter to Prevent Look-Ahead Bias and Dangerous Code
# =====================================================================
class CodeSafetyLinter(ast.NodeVisitor):
def __init__(self):
self.is_safe = True
self.errors = []

def visit_Call(self, node):
# Look for shift() calls with negative parameters (look-ahead)
if isinstance(node.func, ast.Attribute):
if node.func.attr == 'shift':
for arg in node.args:
if isinstance(arg, ast.UnaryOp) and isinstance(arg.op, ast.USub):
if isinstance(arg.operand, ast.Constant) and arg.operand.value > 0:
self.is_safe = False
self.errors.append("Look-ahead bias detected: negative shift parameter used.")
elif isinstance(arg, ast.Constant) and isinstance(arg.value, int) and arg.value < 0:
self.is_safe = False
self.errors.append("Look-ahead bias detected: negative shift parameter used.")

# Prevent calling system processes
if isinstance(node.func, ast.Name):
if node.func.id in ['eval', 'exec', 'open', 'compile']:
self.is_safe = False
self.errors.append(f"Forbidden system-level call: {node.func.id}() is not allowed.")

self.generic_visit(node)

def visit_Import(self, node):
for alias in node.names:
if alias.name in ['os', 'sys', 'subprocess', 'shutil', 'requests']:
self.is_safe = False
self.errors.append(f"Forbidden import: {alias.name} is blocked for security reasons.")
self.generic_visit(node)

def visit_ImportFrom(self, node):
if node.module in ['os', 'sys', 'subprocess', 'shutil', 'requests']:
self.is_safe = False
self.errors.append(f"Forbidden import: {node.module} is blocked for security reasons.")
self.generic_visit(node)

def verify_code_safety(code_str: str) -> Tuple[bool, list]:
try:
tree = ast.parse(code_str)
linter = CodeSafetyLinter()
linter.visit(tree)
return linter.is_safe, linter.errors
except SyntaxError as e:
return False, [f"Syntax error during parsing: {str(e)}"]

# =====================================================================
# 2. Mock Data Generator (For Testing the Agent Loop)
# =====================================================================
def generate_synthetic_data() -> pd.DataFrame:
"""Generates synthetic multi-asset daily data for the backtest engine."""
np.random.seed(42)
dates = pd.date_range(start="2020-01-01", end="2023-12-31", freq="B")
symbols = ["AAPL", "MSFT", "GOOGL", "AMZN", "META"]

records = []
for symbol in symbols:
price = 100.0
for date in dates:
pct_change = np.random.normal(0.0002, 0.015) # slight upward drift
price *= (1 + pct_change)
volume = np.random.randint(100000, 5000000)
records.append({
"date": date,
"symbol": symbol,
"close": price,
"open": price * (1 + np.random.normal(0, 0.002)),
"high": price * (1 + abs(np.random.normal(0, 0.005))),
"low": price * (1 – abs(np.random.normal(0, 0.005))),
"volume": volume
})

df = pd.DataFrame(records)
df.set_index(["date", "symbol"], inplace=True)
return df.sort_index()

# =====================================================================
# 3. Dynamic Execution Sandbox
# =====================================================================
class ExecutionSandbox:
def __init__(self, data: pd.DataFrame):
self.data = data

def execute_strategy(self, code_str: str) -> Dict[str, Any]:
"""Runs generated code in an isolated environment against backtest data."""
safe, errors = verify_code_safety(code_str)
if not safe:
return {"success": False, "error": f"Linter validation failed: {'; '.join(errors)}"}

# Define a controlled local scope
local_scope = {
"pd": pd,
"np": np,
"df": self.data.copy()
}

try:
# We expect the code to define a compute_signals(df) function
exec(code_str, local_scope, local_scope)

if "compute_signals" not in local_scope:
return {"success": False, "error": "Function 'compute_signals(df)' was not defined in the code."}

# Execute the generated strategy function
result_df = local_scope["compute_signals"](self.data.copy())

# Basic validation of returned structure
if not isinstance(result_df, pd.DataFrame):
return {"success": False, "error": "compute_signals must return a pandas DataFrame."}
if "signal" not in result_df.columns:
return {"success": False, "error": "Returned DataFrame is missing the required 'signal' column."}

return {"success": True, "data": result_df}

except Exception as e:
exc_type, exc_value, exc_tb = sys.exc_info()
tb_lines = traceback.format_exception(exc_type, exc_value, exc_tb)
cleaned_traceback = "".join(tb_lines[–4:]) # Last few lines of the error trace
return {
"success": False,
"error": f"Runtime exception: {str(e)}\nTraceback:\n{cleaned_traceback}"
}

# =====================================================================
# 4. Evaluator Engine (Backtest Performance & Slippage)
# =====================================================================
def run_backtest_evaluation(df_signals: pd.DataFrame, transaction_costs_bps: float = 5.0) -> Dict[str, Any]:
"""Calculates quantitative performance metrics from signals with friction."""
try:
# Avoid modifications to original frame
df = df_signals.copy()

# Ensure 'signal' values are constrained to [-1, 1] range
df['signal'] = df['signal'].clip(–1, 1).fillna(0)

# Calculate asset returns
df['next_ret'] = df.groupby('symbol')['close'].pct_change().shift(–1) # return from t to t+1

# Compute dynamic position size changes to calculate transaction costs
df['prev_signal'] = df.groupby('symbol')['signal'].shift(1).fillna(0)
df['trades'] = (df['signal'] – df['prev_signal']).abs()

# Raw Strategy returns (without friction)
df['raw_returns'] = df['signal'] * df['next_ret']

# Friction penalty applied to trades (slippage + commissions)
friction = df['trades'] * (transaction_costs_bps / 10000.0)
df['net_returns'] = df['raw_returns'] – friction

# Group metrics by date to calculate portfolio returns (equal weight per active position)
portfolio_returns = df.groupby('date')['net_returns'].mean()

# Metric Calculations
total_return = (1 + portfolio_returns).prod() – 1
sharpe = 0.0
if portfolio_returns.std() > 0:
sharpe = (portfolio_returns.mean() / portfolio_returns.std()) * np.sqrt(252)

cum_ret = (1 + portfolio_returns).cumprod()
running_max = cum_ret.cummax()
drawdown = (cum_ret – running_max) / running_max
max_drawdown = drawdown.min()

# Metric Payload
return {
"success": True,
"sharpe_ratio": float(np.round(sharpe, 3)),
"max_drawdown": float(np.round(max_drawdown, 4)),
"total_return": float(np.round(total_return, 4)),
"annualized_volatility": float(np.round(portfolio_returns.std() * np.sqrt(252), 4)),
"turnover_rate": float(np.round(df['trades'].mean(), 4))
}
except Exception as e:
return {"success": False, "error": f"Evaluation engine failed: {str(e)}"}

# =====================================================================
# 5. The Agent Coordinator / LLM Integration
# =====================================================================
class QuantAgentOrchestrator:
def __init__(self, api_key: str, data: pd.DataFrame):
self.client = openai.OpenAI(api_key=api_key)
self.sandbox = ExecutionSandbox(data)

def _call_llm(self, system_prompt: str, user_prompt: str) -> str:
response = self.client.chat.completions.create(
model="gpt-4-turbo",
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.1 # Low variance for programmatic generation
)
return response.choices[0].message.content

def run_research_loop(self, hypothesis: str, max_iterations: int = 3) -> Dict[str, Any]:
print(f"[Orchestrator] Starting optimization loop for hypothesis: {hypothesis}\n")

system_coder_prompt = """
You are an elite quantitative analyst and senior pandas engineer.
Your task is to write a python function `compute_signals(df)` that computes a trading signal.
The dataframe passed in is a multi-index DataFrame with index level 0 as 'date' and level 1 as 'symbol'.
Columns present: ['open', 'high', 'low', 'close', 'volume']

CRITICAL RULES:
1. Return the original dataframe with a new column named 'signal' representing position size (values must range between -1.0 and 1.0).
2. Do NOT use positive or negative look-ahead shifts. Absolutely NO `shift(-N)` with N > 0.
3. Avoid loop-based execution. Implement clean, vectorized operations using Pandas/Numpy grouped by symbol.
4. Do NOT import OS, sys, or execute external tasks.
5. Output ONLY clean executable python code inside a single “`python code block. No explanations, no markdown around the block, and no other text.
"""

user_prompt = f"Implement this alpha hypothesis: {hypothesis}"
feedback = ""

for iteration in range(1, max_iterations + 1):
print(f"— Iteration {iteration} of {max_iterations} —")

# Step 1: Request Code from Agent
full_user_prompt = user_prompt
if feedback:
full_user_prompt += f"\n\nYour previous code failed execution or evaluation. Here is the feedback:\n{feedback}\nPlease fix the code and try again."

raw_response = self._call_llm(system_coder_prompt, full_user_prompt)

# Extract python code block
code = self._clean_llm_code_output(raw_response)

print("[Sandbox] Code extracted. Validating syntax and running execution sandbox…")
# Step 2: Sandbox execution
sandbox_result = self.sandbox.execute_strategy(code)

if not sandbox_result["success"]:
print(f"[Sandbox] Failed! Feedback sent to developer.")
feedback = f"Sandbox compilation/execution error:\n{sandbox_result['error']}"
continue

print("[Sandbox] Success! Proceeding to backtest valuation…")

# Step 3: Performance evaluation
eval_metrics = run_backtest_evaluation(sandbox_result["data"])

if not eval_metrics["success"]:
print("[Evaluator] Failed to evaluate backtest metrics.")
feedback = f"Evaluation Error:\n{eval_metrics['error']}"
continue

print(f"[Evaluator] Completed. Sharpe Ratio: {eval_metrics['sharpe_ratio']}, Max Drawdown: {eval_metrics['max_drawdown']}")

# Step 4: Critique and validation check (e.g., Rejecting sub-standard performance)
if eval_metrics["sharpe_ratio"] < 1.0:
print(f"[Critic] Rejected due to low performance (Sharpe {eval_metrics['sharpe_ratio']} < 1.0). Requesting iteration…")
feedback = (
f"The code ran successfully, but the performance is poor:\n"
f"Sharpe Ratio: {eval_metrics['sharpe_ratio']}\n"
f"Max Drawdown: {eval_metrics['max_drawdown']}\n"
f"Total Return: {eval_metrics['total_return']}\n"
f"Please refine the model's coefficients or logic to improve performance without introducing look-ahead bias."
)
continue

print(f"\n[Orchestrator] Strategy successfully optimized on Iteration {iteration}!")
return {
"status": "success",
"code": code,
"metrics": eval_metrics
}

return {
"status": "failed",
"last_feedback": feedback
}

def _clean_llm_code_output(self, raw_text: str) -> str:
"""Parses and strips code blocks from LLM output markdown."""
if "“`python" in raw_text:
parts = raw_text.split("“`python")
code = parts[1].split("“`")[0]
return code.strip()
elif "“`" in raw_text:
parts = raw_text.split("“`")
code = parts[1].split("“`")[0]
return code.strip()
return raw_text.strip()

# =====================================================================
# 6. Execution Script
# =====================================================================
if __name__ == "__main__":
# 1. Setup mock multi-asset dataset
print("[Setup] Generating synthetic tick/daily dataset…")
market_data = generate_synthetic_data()

# 2. Extract API Key (Make sure to set this in your environment)
api_key = os.getenv("OPENAI_API_KEY")

if not api_key:
print("[System Check] OPENAI_API_KEY environment variable not set.")
print("[System Check] Defaulting to execution testing using a pre-saved mock response simulation.")
# Simulating safe run logic locally when no API key is set
dummy_code = """
def compute_signals(df):
# Mean reversion strategy using 20-day rolling z-score of close price
# We group by symbol to avoid leakage between assets
grouped = df.groupby('symbol')
mean = grouped['close'].transform(lambda x: x.rolling(20).mean())
std = grouped['close'].transform(lambda x: x.rolling(20).std())
z_score = (df['close'] – mean) / std

# Generate long/short signals
df['signal'] = 0.0
df.loc[z_score > 1.5, 'signal'] = -1.0
df.loc[z_score < -1.5, 'signal'] = 1.0
return df
"""
sandbox = ExecutionSandbox(market_data)
sandbox_res = sandbox.execute_strategy(dummy_code)
if sandbox_res["success"]:
metrics = run_backtest_evaluation(sandbox_res["data"])
print(f"[Dry Run Metrics] Simulated execution Sharpe: {metrics['sharpe_ratio']}")
else:
# Run live research orchestration loop
orchestrator = QuantAgentOrchestrator(api_key=api_key, data=market_data)

hypothesis = (
"Mean reversion strategy. Compute 20-day rolling average and standard deviation. "
"Go short when the price crosses above 1.5 standard deviations above the mean, "
"and long when it falls below 1.5 standard deviations. Limit maximum position sizes to 1."
)

result = orchestrator.run_research_loop(hypothesis=hypothesis, max_iterations=3)
print("\n=== Final Run Summary ===")
print(json.dumps(result, indent=2))

from scipy.signal import savgol_filter
# Inside compute_signals…
df['signal'] = savgol_filter(df['close'].values, window_length=15, polyorder=2)

df['volatility'] = df.groupby('symbol')['close'].rolling_std(window=20)

Runtime exception: 'SeriesGroupBy' object has no attribute 'rolling_std'
Traceback:
File "sandbox_exec.py", line 12, in compute_signals
df['volatility'] = df.groupby('symbol')['close'].rolling_std(window=20)