Skip to main content
Create custom routines to extend agent capabilities with deterministic Python code.

​
Routine Requirements

Every routine module needs:
  1. Config - A Pydantic BaseModel with parameters (docstring becomes the description)
  2. run(config, context) - Async function that executes the routine and returns a string
  3. CONTINUOUS = True (optional) - Mark as continuous routine with internal loop

​
Create a Custom Routine

​
1. Create the File

In your agent’s routines/ folder: 
# trading_agents/sol_scalper/routines/momentum_scanner.py
from pydantic import BaseModel, Field
from telegram.ext import ContextTypes

from config_manager import get_client


class Config(BaseModel):
    """Scan for momentum breakouts on a trading pair."""

    connector_name: str = Field(default="binance_perpetual", description="Exchange connector")
    trading_pair: str = Field(default="SOL-USDT", description="Market to analyze")
    lookback: int = Field(default=20, description="Number of candles")


async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    """Scan for momentum breakouts."""
    chat_id = context._chat_id if hasattr(context, "_chat_id") else None
    client = await get_client(chat_id, context=context)

    if not client:
        return "No server available"

    # Get candles via Hummingbot API
    candles = await client.market_data.get_candles(
        connector_name=config.connector_name,
        trading_pair=config.trading_pair,
        interval="1m",
        limit=config.lookback
    )

    # Calculate momentum
    closes = [c["close"] for c in candles["candles"]]
    momentum = (closes[-1] - closes[0]) / closes[0] * 100

    # Volume confirmation
    volumes = [c["volume"] for c in candles["candles"]]
    avg_volume = sum(volumes) / len(volumes)
    volume_surge = volumes[-1] > avg_volume * 1.5

    # Generate signal
    if momentum > 2 and volume_surge:
        signal = "🟢 STRONG BULLISH"
    elif momentum > 1:
        signal = "🟡 Bullish"
    elif momentum < -2 and volume_surge:
        signal = "🔴 STRONG BEARISH"
    elif momentum < -1:
        signal = "🟠 Bearish"
    else:
        signal = "⚪ Neutral"

    return (
        f"**Momentum Scanner** - {config.trading_pair}\n"
        f"Signal: {signal}\n"
        f"Momentum: {momentum:+.2f}%\n"
        f"Volume surge: {'Yes' if volume_surge else 'No'}\n"
        f"Price: ${closes[-1]:,.2f}"
    )

​
2. Register the Routine

Add to your agent’s config or the routine will be auto-discovered.

​
3. Use in Agent

Update your agent.md: 
## Entry Rules

1. Run momentum_scanner routine
2. If signal is "strong_bullish" → Open long
3. If signal is "strong_bearish" → Open short
4. Otherwise → Wait

​
4. Test

Run a dry run to verify: 
/agent → SOL Scalper → Dry Run

Dry Run Result:
- Ran momentum_scanner
- Result: {signal: "strong_bullish", momentum: 2.5%}
- Decision: Would open LONG

​
Creating with Condor

You can ask Condor to create routines for you via natural language: 
You: Create a routine that analyzes support and resistance levels
     using EMAs of 7, 25, and 99 on 1-minute candles.
Condor generates the routine: 
# routines/support_resistance_ema_levels.py
from pydantic import BaseModel, Field
from telegram.ext import ContextTypes

from config_manager import get_client


class Config(BaseModel):
    """Analyze support/resistance levels and EMA alignment."""

    connector_name: str = Field(default="binance_perpetual", description="Exchange connector")
    trading_pair: str = Field(default="BTC-USDT", description="Trading pair")
    candle_count: int = Field(default=100, description="Number of candles to analyze")
    lookback: int = Field(default=20, description="Lookback for S/R detection")


async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    """Identify support/resistance levels and EMA alignment."""
    chat_id = context._chat_id if hasattr(context, "_chat_id") else None
    client = await get_client(chat_id, context=context)

    candles = await client.market_data.get_candles(
        connector_name=config.connector_name,
        trading_pair=config.trading_pair,
        interval="1m",
        limit=config.candle_count
    )

    closes = [c["close"] for c in candles["candles"]]

    # Calculate EMAs
    ema_7 = calculate_ema(closes, 7)
    ema_25 = calculate_ema(closes, 25)
    ema_99 = calculate_ema(closes, 99)

    # Determine trend from EMA alignment
    if ema_7 > ema_25 > ema_99:
        trend = "🟢 Bullish"
    elif ema_7 < ema_25 < ema_99:
        trend = "🔴 Bearish"
    else:
        trend = "🟡 Mixed"

    # Find support/resistance levels
    supports, resistances = find_sr_levels(closes, config.lookback)

    return (
        f"**EMA Analysis** - {config.trading_pair}\n"
        f"Trend: {trend}\n"
        f"EMA 7: ${ema_7:,.2f}\n"
        f"EMA 25: ${ema_25:,.2f}\n"
        f"EMA 99: ${ema_99:,.2f}\n"
        f"Support: ${supports[0]:,.2f}\n"
        f"Resistance: ${resistances[0]:,.2f}"
    )
The routine is auto-discovered and immediately available via /routines.

​
Routine Best Practices

Same input should always produce same output. Avoid randomness or time-based logic that would make results unpredictable.
Return human-readable strings for Telegram display. Use markdown formatting for clarity.
Return error messages rather than raising exceptions:
if not client:
    return "No server available"
if not candles:
    return "No candle data available"
The Config class docstring becomes the routine description in the UI. Use Field(description=...) to document each parameter.
Continuous routines must catch asyncio.CancelledError to clean up gracefully when stopped.

​
Global vs Agent-Specific

LocationScope
~/condor/routines/Available to all agents
trading_agents/{slug}/routines/Specific to one agent
Agent-specific routines override global ones with the same name.