Building a trading signal agent with CrewAI

A trading signal agent crewai project needs more than a single prompt loop—it needs separated concerns for data, analysis, and risk. This tutorial builds a three-agent CrewAI crew that fetches daily OHLCV data, identifies momentum shifts, and outputs a constrained signal you can wire into a broker.

Prerequisites

Python 3.10 or newer
Installed packages: crewai, yfinance, langchain-openai, openai, pandas
An OpenAI-compatible API key (any provider works)

pip install crewai yfinance langchain-openai openai pandas

Set your key in the environment:

export OPENAI_API_KEY="sk-..."

Project structure

Keep it flat for a tutorial. One file signal_crew.py is enough.

import os
from langchain_openai import ChatOpenAI
from crewai import Agent, Task, Crew, Process, tool
import yfinance as yf

Configuring the LLM

CrewAI accepts any LangChain chat model. We use ChatOpenAI so we can point at any OpenAI-compatible base URL. Swap the base_url if you run a gateway.

llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.2,
    base_url="https://api.openai.com/v1"  # or your inference gateway
)

If you’d rather not juggle multiple provider keys, point that client at n4n.ai’s single OpenAI-compatible endpoint—it fronts 240+ models and fails over automatically when a provider is degraded.

Building a data tool

LLMs shouldn’t hallucinate numbers. Give the data agent a real function that pulls from Yahoo Finance and computes two standard indicators.

@tool("Fetch OHLCV and indicators")
def fetch_indicators(ticker: str) -> str:
    """Pull 30 days of daily bars and return price, 20-day SMA, 14-day RSI."""
    df = yf.download(ticker, period="30d", interval="1d", progress=False)
    if df.empty:
        return "{}"
    close = df["Close"].squeeze()
    sma20 = close.rolling(20).mean().iloc[-1]
    delta = close.diff()
    gain = delta.clip(lower=0).rolling(14).mean()
    loss = -delta.clip(upper=0).rolling(14).mean()
    rs = gain / loss
    rsi14 = (100 - (100 / (1 + rs))).iloc[-1]
    return str({
        "ticker": ticker,
        "price": round(float(close.iloc[-1]), 2),
        "sma20": round(float(sma20), 2),
        "rsi14": round(float(rsi14), 2),
    })

Expected output when called with "AAPL":

{'ticker': 'AAPL', 'price': 227.34, 'sma20': 222.10, 'rsi14': 58.41}

Defining the agents

The trading signal agent crewai design splits labor: one agent gets data, one forms a view, one enforces risk.

data_agent = Agent(
    role="Market Data Engineer",
    goal="Retrieve clean OHLCV data and compute basic indicators",
    backstory="Expert at pulling and conditioning financial time series.",
    llm=llm,
    tools=[fetch_indicators],
    verbose=True,
)

strategy_agent = Agent(
    role="Quant Strategist",
    goal="Detect momentum and mean-reversion signals from indicators",
    backstory="Former hedge fund analyst specializing in short-horizon signals.",
    llm=llm,
    verbose=True,
)

risk_agent = Agent(
    role="Risk Officer",
    goal="Validate signals against position limits and volatility caps",
    backstory="Disciplined risk manager who kills trades that exceed risk budget.",
    llm=llm,
    verbose=True,
)

Defining tasks

Tasks declare what each agent must produce. Use context to chain them.

fetch_task = Task(
    description="Call the indicator tool for {ticker}. Return the JSON unchanged.",
    expected_output="Raw indicator JSON.",
    agent=data_agent,
)

analyze_task = Task(
    description="Read the JSON. Decide LONG if price>sma20 and rsi14 30-70, "
                "SHORT if price<sma20 and rsi14 30-70, else FLAT. Give confidence 0-1.",
    expected_output="Signal: LONG/SHORT/FLAT, confidence, rationale.",
    context=[fetch_task],
    agent=strategy_agent,
)

risk_task = Task(
    description="Reject signal if confidence<0.6. Output final action and max size 2% NAV.",
    expected_output="Final signal with size limit or rejection reason.",
    context=[analyze_task],
    agent=risk_agent,
)

Assembling and running the crew

crew = Crew(
    agents=[data_agent, strategy_agent, risk_agent],
    tasks=[fetch_task, analyze_task, risk_task],
    process=Process.sequential,
    verbose=True,
)

result = crew.kickoff(inputs={"ticker": "AAPL"})
print("FINAL:", result)

A sample run log (truncated):

[Data Agent] Invoking tool: Fetch OHLCV and indicators
[Strategy Agent] Price 227.34 > SMA 222.10, RSI 58.41 -> LONG, conf 0.72
[Risk Agent] Confidence 0.72 >= 0.6 -> APPROVED. Size 2% NAV.
FINAL: LONG AAPL, size 2% NAV, confidence 0.72

Extending the trading signal agent crewai with live checks

The above runs once. In production you want a loop and guardrails:

Cache the indicator pull for 15 minutes; don’t hit Yahoo every request.
Log each agent’s raw output for audit.
Strip the final string and parse via json.loads if you need structured execution.

import time, json, redis

r = redis.Redis()

def get_cached(ticker):
    cached = r.get(f"ind:{ticker}")
    if cached:
        return cached
    raw = fetch_indicators.run(ticker)
    r.setex(f"ind:{ticker}", 900, raw)
    return raw

Wire get_cached into the tool body to cut LLM waits and external rate limits.

Why separate agents instead of one prompt

A single prompt that asks for “data, analysis, and risk” leaks context and produces inconsistent JSON. Splitting forces the strategist to reason only from the data agent’s structured output, and the risk agent to reason only from the strategist’s signal. You can swap the strategist model for a cheaper one and keep the risk agent on a stronger model—CrewAI lets you set llm per agent.

Closing notes for production

Treat the output as advisory. Before sending an order, verify the signal against your own position server and compliance rules. If you run this at scale, use per-token metering and provider routing directives to control cost; the crew’s LLM calls are just standard chat completions under the hood. The trading signal agent crewai pattern here is a skeleton—add a news sentiment agent or a macro context task before the strategist when you need richer signals.