create-routine

创建 / 编辑例行任务

你正在为 Condor 编写例行任务——这是一个从 routines/ 自动发现的 Python 脚本。例行任务通过 Telegram（/routines）或 Web 仪表板运行。

不是 Agent 例行任务。 Agent 例行任务位于交易 Agent 策略内，并通过 /trading-agent-builder 创建。

最简例行任务

from pydantic import BaseModel, Field
from telegram.ext import ContextTypes
from config_manager import get_client

CATEGORY = "Market Data"  # Market Data | Analysis | Arbitrage | Monitoring | Bot Analysis

class Config(BaseModel):
    """One-line description shown in UI."""
    trading_pair: str = Field(default="BTC-USDT", description="Trading pair")
    connector_name: str = Field(default="binance_perpetual", description="Exchange connector")

async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    client = await get_client(context._chat_id, context=context)
    if not client:
        return "No server available"
    # ... do work ...
    return "result string"

主要规则

• 文件放在 routines/ 中，命名为 snake_case.py
• 必须导出 Config（Pydantic BaseModel）和 async def run(config, context) -> str
• Config.__doc__ = UI 中显示的例行任务描述
• 模块级别的 CATEGORY 用于在目录中对例行任务分组
• 返回字符串，或用于丰富输出的 RoutineResult
• get_client() 是可选的——例行任务可以直接使用外部 API（aiohttp 等）
• 使用 asyncio.gather 并行获取数据
• 妥善处理缺失数据——返回错误字符串，不要抛出异常

丰富输出

from routines.base import RoutineResult

# Tables in web dashboard
return RoutineResult(
    text="Summary for Telegram",
    table_data=[{"Pair": "BTC-USDT", "Price": 100000}],
    table_columns=["Pair", "Price"],
)

# Chart image sent to Telegram
return RoutineResult(text=summary, chart_image=png_bytes)

# KPI cards in web dashboard
return RoutineResult(text=summary, sections=[
    {"type": "kpi", "label": "Price", "value": "$100K", "delta": "+5%", "trend": "up"},
])

ReportBuilder（HTML 报告）

始终在 try/except 内延迟导入：

try:
    from condor.reports import ReportBuilder
    builder = ReportBuilder("Report Title")
    builder.source("routine", "routine_name").tags(["tag1", "tag2"])
    builder.kpi("Price", "$100K", delta="+5%", trend="up")  # individual calls, NOT a list
    builder.markdown("## Analysis\nSome text")                # use markdown() for all text/headings
    builder.table([{"Col": "val"}])                           # columns auto-detected from first row
    builder.plotly(fig)                                        # Plotly figure object
    builder.manual_order()                                     # preserve insertion order (default: kpi→plotly→table→markdown)
    builder.save()
except Exception as e:
    logger.warning(f"Report generation failed: {e}")

仅有以下方法： source、tags、kpi、markdown、table、plotly、manual_order、save。没有 heading()、text()、section() 或 html()。

连续例行任务的实时报告

对于每个周期都会更新报告的连续例行任务，请使用 LiveReport：

from condor.reports import LiveReport

report = LiveReport("Monitor Title", source_name="routine_name", tags=["live"])
history = []

try:
    while True:
        # ... fetch data ...
        history.append({"Time": now, "Price": price})

        report.clear()  # reset builder for fresh render
        report.builder.manual_order()
        report.builder.kpi("Price", f"${price:,.2f}")
        report.builder.table(history[-50:])
        report.update()  # creates on first call, updates thereafter

        await asyncio.sleep(interval)
except asyncio.CancelledError:
    return "Stopped"

LiveReport API： clear()、update()、report_id（属性）、builder（属性——底层的 ReportBuilder）

执行上下文

例行任务会在 3 种不同的上下文中运行——代码必须在所有上下文中正常工作：

上下文	context.bot	context._chat_id	触发方式
Telegram	真实 bot（python-telegram-bot）	用户的聊天 ID	/routines 命令
Web Dashboard	_HttpBot（HTTP 回退）	用户 ID 或 0	Web API
MCP	_HttpBot（HTTP 回退）	settings.chat_id 或 0	manage_routines 工具

要点： context.bot 始终可用——永远不会是 None。在非 Telegram 上下文中，它是一个 _HttpBot，通过使用 TELEGRAM_TOKEN 的 Telegram HTTP API 发送消息。你始终可以安全地调用 context.bot.send_message(...)。

_HttpBot 支持的功能

• send_message(chat_id=..., text=..., parse_mode=...)
• send_photo(chat_id=..., photo=..., caption=...)
• send_document(chat_id=..., document=..., caption=...)
• edit_message_text(chat_id=..., message_id=..., text=...)

如果未设置 TELEGRAM_TOKEN，调用会被静默忽略（不会崩溃）。

连续例行任务

对于包含内部循环的例行任务，设置 CONTINUOUS = True。这类例行任务会作为 asyncio 任务运行，直到被取消。

import asyncio
from pydantic import BaseModel, Field
from telegram.ext import ContextTypes
from config_manager import get_client

CONTINUOUS = True

class Config(BaseModel):
    """Live price monitor with alerts."""
    connector: str = Field(default="binance", description="Exchange connector")
    trading_pair: str = Field(default="BTC-USDT", description="Trading pair")
    threshold_pct: float = Field(default=1.0, description="Alert threshold %")
    interval_sec: int = Field(default=10, description="Check interval in seconds")

async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    chat_id = context._chat_id
    client = await get_client(chat_id, context=context)
    if not client:
        return "No server available"

    # Send start notification (works in all contexts)
    await context.bot.send_message(
        chat_id=chat_id,
        text=f"Started monitoring {config.trading_pair}",
    )

    last_price = None
    try:
        while True:
            prices = await client.market_data.get_prices(
                connector_name=config.connector,
                trading_pairs=config.trading_pair,
            )
            current = prices["prices"].get(config.trading_pair)

            if current and last_price:
                change = abs((current - last_price) / last_price) * 100
                if change >= config.threshold_pct:
                    await context.bot.send_message(
                        chat_id=chat_id,
                        text=f"Alert: {config.trading_pair} moved {change:.2f}%",
                    )
            last_price = current or last_price
            await asyncio.sleep(config.interval_sec)

    except asyncio.CancelledError:
        return "Stopped"

连续例行任务规则：

• 始终在外层循环中捕获 asyncio.CancelledError——重新抛出异常或返回
• 使用 context.bot.send_message() 发送实时通知（在所有上下文中均可用）
• 内层循环的异常应捕获并记录日志，不要重新抛出
• 取消时返回摘要字符串

向 Telegram 发送图表

buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150)  # matplotlib
# OR: fig.write_image(buf, format="png", scale=2)  # plotly
buf.seek(0)

# Works in all contexts (Telegram, Web, MCP)
await context.bot.send_photo(chat_id=context._chat_id, photo=buf, caption="Title")

# Also return as RoutineResult for web dashboard
return RoutineResult(text=summary, chart_image=buf.getvalue())

Hummingbot 客户端 API

client = await get_client(context._chat_id, context=context)

# Market data
await client.market_data.get_candles(connector, pair, interval="1m", max_records=100)
await client.market_data.get_order_book(connector, pair, depth=10)
await client.market_data.get_prices(connector, trading_pairs)           # str or list
await client.market_data.get_funding_info(connector, pair)
await client.market_data.get_price_for_volume(connector, pair, volume, is_buy)
await client.market_data.get_historical_candles(connector, pair, interval, start_time, end_time)
await client.market_data.get_candles_last_days(connector, pair, days, interval="1h")

# Portfolio
await client.portfolio.get_state(account_names=None, connector_names=None)
await client.portfolio.get_total_value()  # returns float
await client.portfolio.get_distribution()
await client.portfolio.get_history(limit=100, interval=None)

# Executors
await client.executors.search_executors(controller_ids=[], status="active", limit=50)
await client.executors.get_performance_report(controller_id=cid)  # NOT executor_id
await client.executors.create_executor(executor_config_dict)

解析响应

# Candles — handle both formats
result = await client.market_data.get_candles(connector, pair, interval="1m", max_records=100)
records = result if isinstance(result, list) else result.get("data", result.get("candles", []))

# Order book
ob = await client.market_data.get_order_book(connector, pair, depth=10)
bids, asks = ob.get("bids", []), ob.get("asks", [])  # [[price, size], ...]

# Bounded concurrency for bulk fetches
sem = asyncio.Semaphore(10)
async def fetch(p):
    async with sem:
        return await client.market_data.get_candles(connector, p, interval="1m", max_records=100)
results = await asyncio.gather(*[fetch(p) for p in pairs], return_exceptions=True)

Plotly 图表规则

• 图例始终置于底部： 每个 Plotly 图表都必须设置 fig.update_layout(legend=dict(orientation="h", yanchor="top", y=-0.15, xanchor="center", x=0.5))，以确保图例水平显示在图表下方，而不是顶部或侧边。

常见错误

• 使用 get_order_book()，不要使用 get_order_book_snapshot
• 使用 get_candles(connector, pair, interval, max_records)，不要使用 get_candles(pair, interval, limit)
• 使用 get_performance_report(controller_id=...)，不要使用 get_performance_report(executor_id=...)
• create_executor(config_dict) — 使用普通 dict，不要使用 Pydantic 模型
• builder.kpi(label, value) — 传入单独的参数，不要传入 dict 列表
• 所有客户端方法都是异步的——始终使用 await
• get_total_value() 返回 float，其他方法均返回 dict