create-routine

What this skill does

# 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`.

## 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"
```

## 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

## Rich Output

```python
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 Reports)

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()`.

### 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`)

## 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.

### What `_HttpBot` supports
- `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=...)`

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

## 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 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

## Sending Charts to Telegram

```python
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 Client API

```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, p