create-routine

Compare original and translation side by side

🇺🇸

Original

English

🇨🇳

Translation

Chinese

Create / Edit Routine

创建/编辑程序

You are working on a routine for Condor — a Python script auto-discovered from

routines/

. Routines run via Telegram (

/routines

) or the web dashboard.

Not agent routines. Agent routines live inside trading agent strategies and are created via
/trading-agent-builder
.

你正在为Condor开发一个程序——这是一个从

routines/

自动发现的Python脚本。程序可通过Telegram（

/routines

命令）或Web控制面板运行。

注意：不是Agent程序。Agent程序位于交易Agent策略内部，需通过
/trading-agent-builder
创建。

Minimal Routine

最小化程序示例

python

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"

python

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):
    """UI中显示的单行描述。"""
    trading_pair: str = Field(default="BTC-USDT", description="交易对")
    connector_name: str = Field(default="binance_perpetual", description="交易所连接器")

async def run(config: Config, context: ContextTypes.DEFAULT_TYPE) -> str:
    client = await get_client(context._chat_id, context=context)
    if not client:
        return "无可用服务器"
    # ... 执行任务 ...
    return "结果字符串"

Key Rules

核心规则

File goes in
```
routines/
```
as
```
snake_case.py
```

Must export

Config

(Pydantic BaseModel) and

async def run(config, context) -> str

```
Config.__doc__
```
= routine description in UI
```
CATEGORY
```
at module level groups it in the catalog
Return a string, or
```
RoutineResult
```
for rich output
```
get_client()
```
is optional — routines can use external APIs directly (aiohttp, etc.)
Use
```
asyncio.gather
```
for parallel fetches
Handle missing data gracefully — return error strings, don't raise

文件需以
```
snake_case.py
```
命名并存放在
```
routines/
```
目录下
必须导出
```
Config
```
（Pydantic BaseModel）和
```
async def run(config, context) -> str
```
函数
```
Config.__doc__
```
为程序在UI中显示的描述文本
模块级别的
```
CATEGORY
```
用于在目录中对程序进行分组
返回字符串，或使用
```
RoutineResult
```
实现富文本输出
```
get_client()
```
为可选依赖——程序可直接调用外部API（如aiohttp等）
使用
```
asyncio.gather
```
实现并行数据获取
优雅处理缺失数据——返回错误字符串，而非抛出异常

Rich Output

富文本输出

python

from routines.base import RoutineResult

python

from routines.base import RoutineResult

Tables in web dashboard

Web控制面板中的表格

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

return RoutineResult( text="Telegram端摘要", table_data=[{"Pair": "BTC-USDT", "Price": 100000}], table_columns=["Pair", "Price"], )

Chart image sent to Telegram

发送至Telegram的图表图片

return RoutineResult(text=summary, chart_image=png_bytes)

KPI cards in web dashboard

Web控制面板中的KPI卡片

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

undefined

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

undefined

ReportBuilder (HTML Reports)

ReportBuilder（HTML报告生成）

Always lazy-import inside try/except:

python

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}")

Only these methods exist:

source

tags

kpi

markdown

table

plotly

manual_order

save

. No

heading()

text()

section()

, or

html()

请始终在try/except块中延迟导入：

python

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")  # 单独调用，而非传入列表
    builder.markdown("## Analysis\nSome text")                # 使用markdown()处理所有文本/标题
    builder.table([{"Col": "val"}])                           # 自动从第一行检测列名
    builder.plotly(fig)                                        # Plotly图形对象
    builder.manual_order()                                     # 保留插入顺序（默认顺序：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()

。

Live Reports for Continuous Routines

持续运行程序的实时报告

Use

LiveReport

for continuous routines that produce a living report updated each tick:

python

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

(property),

builder

(property — the underlying

ReportBuilder

)

对于需持续运行并生成实时更新报告的程序，请使用

LiveReport

：

python

from condor.reports import LiveReport

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

try:
    while True:
        # ... 获取数据 ...
        history.append({"Time": now, "Price": price})

        report.clear()  # 重置生成器以重新渲染
        report.builder.manual_order()
        report.builder.kpi("Price", f"${price:,.2f}")
        report.builder.table(history[-50:])
        report.update()  # 首次调用创建报告，后续调用更新报告

        await asyncio.sleep(interval)
except asyncio.CancelledError:
    return "已停止"

LiveReport API：

clear()

update()

report_id

（属性）,

builder

（属性——底层的

ReportBuilder

）

Execution Contexts

执行上下文

Routines run in 3 different contexts — your code must work in all of them:

Context	`context.bot`	`context._chat_id`	Trigger
Telegram	Real bot (python-telegram-bot)	User's chat ID	`/routines` command
Web Dashboard	`_HttpBot` (HTTP fallback)	User ID or 0	Web API
MCP	`_HttpBot` (HTTP fallback)	`settings.chat_id` or 0	`manage_routines` tool

Key point:

context.bot

is always available — never

None

. In non-Telegram contexts, it's an

_HttpBot

that sends messages via the Telegram HTTP API using

TELEGRAM_TOKEN

. You can always call

context.bot.send_message(...)

safely.

程序会在3种不同上下文中运行——你的代码必须适配所有场景：

上下文	`context.bot`	`context._chat_id`	触发方式
Telegram	真实机器人（python-telegram-bot）	用户聊天ID	`/routines` 命令
Web控制面板	`_HttpBot` （HTTP降级方案）	用户ID或0	Web API
MCP	`_HttpBot` （HTTP降级方案）	`settings.chat_id` 或0	`manage_routines` 工具

核心要点：

context.bot

始终可用——永远不会为

None

。在非Telegram上下文中，它是

_HttpBot

，通过Telegram HTTP API（使用

TELEGRAM_TOKEN

）发送消息。你可以安全地调用

context.bot.send_message(...)

。

What

_HttpBot

supports

_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

is not set, calls are silently ignored (no crash).

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 Routines

持续运行程序

Set

CONTINUOUS = True

for routines with internal loops. These run as asyncio tasks until cancelled.

python

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"

对于包含内部循环的程序，设置

CONTINUOUS = True

。这类程序会以asyncio任务的形式运行，直到被取消。

python

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

CONTINUOUS = True

class Config(BaseModel):
    """带告警功能的实时价格监控器。"""
    connector: str = Field(default="binance", description="交易所连接器")
    trading_pair: str = Field(default="BTC-USDT", description="交易对")
    threshold_pct: float = Field(default=1.0, description="告警阈值百分比")
    interval_sec: int = Field(default=10, description="检查间隔（秒）")

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 "无可用服务器"

    # 发送启动通知（适配所有上下文）
    await context.bot.send_message(
        chat_id=chat_id,
        text=f"已开始监控{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"告警：{config.trading_pair}波动{change:.2f}%",
                    )
            last_price = current or last_price
            await asyncio.sleep(config.interval_sec)

    except asyncio.CancelledError:
        return "已停止"

Continuous routine rules:

持续运行程序规则：

Always catch
```
asyncio.CancelledError
```
at the outer loop — re-raise or return
Use
```
context.bot.send_message()
```
for real-time notifications (works in all contexts)
Inner loop exceptions should be caught and logged, NOT re-raised
Return a summary string when cancelled

始终在外层循环捕获
```
asyncio.CancelledError
```
——可重新抛出或返回结果
使用
```
context.bot.send_message()
```
发送实时通知（适配所有上下文）
内层循环异常需捕获并记录日志，而非重新抛出
被取消时返回摘要字符串

Sending Charts to Telegram

向Telegram发送图表

python

buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150)  # matplotlib

python

buf = io.BytesIO()
fig.savefig(buf, format="png", dpi=150)  # matplotlib

OR: fig.write_image(buf, format="png", scale=2) # plotly

或：fig.write_image(buf, format="png", scale=2) # plotly

buf.seek(0)

Works in all contexts (Telegram, Web, MCP)

适配所有上下文（Telegram、Web、MCP）

await context.bot.send_photo(chat_id=context._chat_id, photo=buf, caption="Title")

await context.bot.send_photo(chat_id=context._chat_id, photo=buf, caption="标题")

Also return as RoutineResult for web dashboard

同时返回为RoutineResult供Web控制面板展示

return RoutineResult(text=summary, chart_image=buf.getvalue())

undefined

return RoutineResult(text=summary, chart_image=buf.getvalue())

undefined

Hummingbot Client API

python

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

python

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")

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) # 字符串或列表 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)

await client.portfolio.get_state(account_names=None, connector_names=None) await client.portfolio.get_total_value() # 返回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)

undefined

await client.executors.search_executors(controller_ids=[], status="active", limit=50) await client.executors.get_performance_report(controller_id=cid) # 注意：不是executor_id await client.executors.create_executor(executor_config_dict)

undefined

Parsing responses

响应解析

python

undefined

python

undefined

Candles — handle both formats

K线数据——兼容两种格式

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], ...]

ob = await client.market_data.get_order_book(connector, pair, depth=10) bids, asks = ob.get("bids", []), ob.get("asks", []) # [[价格, 数量], ...]

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)

undefined

undefined

Plotly Chart Rules

Plotly图表规则

Legend always at the bottom: Every Plotly figure must set
```
fig.update_layout(legend=dict(orientation="h", yanchor="top", y=-0.15, xanchor="center", x=0.5))
```
so the legend appears horizontally below the chart, never on top or to the side.

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

Common Mistakes

常见错误

```
get_order_book()
```
NOT ~~get_order_book_snapshot~~

get_candles(connector, pair, interval, max_records)

NOT ~~get_candles(pair, interval, limit)~~

get_performance_report(controller_id=...)

NOT ~~get_performance_report(executor_id=...)~~

```
create_executor(config_dict)
```
— plain dict, NOT Pydantic model
```
builder.kpi(label, value)
```
— individual args, NOT a list of dicts
All client methods are async — always
```
await
```
```
get_total_value()
```
returns
```
float
```
, all others return
```
dict
```

使用
```
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)
```
需传入普通字典，而非Pydantic模型
```
builder.kpi(label, value)
```
需传入单独参数，而非字典列表
所有客户端方法均为异步方法——必须使用
```
await
```
```
get_total_value()
```
返回
```
float
```
类型，其他方法均返回
```
dict
```
类型

create-routine

Original

Translation

Create / Edit Routine

创建/编辑程序

Minimal Routine

最小化程序示例

Key Rules

核心规则

Rich Output

富文本输出

Tables in web dashboard

Web控制面板中的表格

Chart image sent to Telegram

发送至Telegram的图表图片

KPI cards in web dashboard

Web控制面板中的KPI卡片

ReportBuilder (HTML Reports)

ReportBuilder（HTML报告生成）

Live Reports for Continuous Routines

持续运行程序的实时报告

Execution Contexts

执行上下文

What _HttpBot supports

_HttpBot支持的方法

Continuous Routines

持续运行程序

Continuous routine rules:

持续运行程序规则：

Sending Charts to Telegram

向Telegram发送图表

OR: fig.write_image(buf, format="png", scale=2) # plotly

或：fig.write_image(buf, format="png", scale=2) # plotly

Works in all contexts (Telegram, Web, MCP)

适配所有上下文（Telegram、Web、MCP）

Also return as RoutineResult for web dashboard

同时返回为RoutineResult供Web控制面板展示

Hummingbot Client API

Hummingbot Client API

Market data

市场数据

Portfolio

投资组合

Executors

执行器

Parsing responses

响应解析

Candles — handle both formats

K线数据——兼容两种格式

Order book

订单簿

Bounded concurrency for bulk fetches

批量获取时的并发限制

Plotly Chart Rules

Plotly图表规则

Common Mistakes

常见错误

What
`_HttpBot`
supports

`_HttpBot`
支持的方法