create-routine

ルーチンの作成 / 編集

あなたは Condor のルーチンに取り組んでいます。これは routines/ から自動検出される Python スクリプトです。ルーチンは Telegram（/routines）または Web ダッシュボードから実行されます。

エージェントルーチンではありません。 エージェントルーチンはトレーディングエージェントのストラテジー内に存在し、/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	実際のボット（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 コンテキストでは、TELEGRAM_TOKEN を使用して Telegram HTTP API 経由でメッセージを送信する _HttpBot です。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 をキャッチします — 再送出するか return します
• リアルタイム通知には 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() 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) — プレーンな dict、NOT Pydantic モデル
• builder.kpi(label, value) — 個別の引数、NOT dict のリスト
• すべてのクライアントメソッドは async です — 常に await します
• get_total_value() は float を返し、それ以外はすべて dict を返します