aaron/tradingview-alert-dispatcher
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1

Browse Source
This commit is contained in:
aaron 2026-05-14 22:12:31 +08:00
parent d01c0d49cd
commit 4f025c3736
9 changed files with 495 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
.env.example Normal file
View File

@ -0,0 +1,18 @@
APP_HOST=0.0.0.0
APP_PORT=8030
DATABASE_PATH=data/dispatcher.db
ADMIN_USERNAME=admin
ADMIN_PASSWORD=12345678
SESSION_SECRET=DBUwycvdxjSUZX4LMvUKa0xMzWKzFJmg
# Optional but recommended. When set, TradingView must call:
# /webhook/tradingview?token=your-shared-secret
# or send X-Webhook-Token: your-shared-secret
WEBHOOK_TOKEN=vvyVmc33aC0I85LkH4yrd6ojvkqmyrb1
RETENTION_DAYS=30
MAX_DELIVERY_ATTEMPTS=3
RETRY_BACKOFF_SECONDS=60
FEISHU_TIMEOUT_SECONDS=10
WORKER_INTERVAL_SECONDS=15

19
README.md
View File

@ -1,6 +1,6 @@
# TradingView Alert Dispatcher
接收 TradingView webhook alert，按 `timeframe + symbol + strategy` 路由到飞书 webhook，并提供管理控制台。
接收 TradingView webhook alert，按 `timeframe / symbol / strategy` 等条件路由到飞书 webhook，并提供管理控制台。
## Run Locally
@ -44,16 +44,22 @@ POST /webhook/tradingview
Content-Type: application/json
```
如果设置了 `WEBHOOK_TOKEN`，TradingView 需要使用以下任一方式携带 token：
```text
POST /webhook/tradingview?token=your-shared-secret
X-Webhook-Token: your-shared-secret
```
`docker-compose.yml` 默认已经设置了占位 token，生产使用前请替换；本地临时调试如果不想校验 token，可以把 `WEBHOOK_TOKEN` 置为空。
## Feishu Message Templates
路由规则支持两种消息类型：
- `Card`：默认，发送飞书 interactive card。
- `Text`：发送普通文本消息。
路由规则统一发送飞书 interactive card。
标题和正文模板支持 `{{field}}` 占位符，字段来自 TradingView alert JSON。嵌套字段可以写成 `{{order.id}}`。
每条路由规则通过「发送到」下拉框选择一个飞书 Webhook。需要同一个信号发到多个群时，可以建多条匹配条件相同、目标不同的规则，并用优先级控制命中顺序；当前默认路由逻辑只发送最高优先级命中的规则。
每条路由规则通过「发送到」下拉框选择一个飞书 Webhook。`timeframe`、`symbol`、`strategy` 至少填写一个，空字段表示不限。例如只填 `symbol=BTCUSDT` 会匹配所有 BTCUSDT 信号。需要同一个信号发到多个群时，可以建多条匹配条件相同、目标不同的规则，并用优先级控制命中顺序；当前默认路由逻辑只发送最高优先级命中的规则。
示例正文模板：
@ -70,6 +76,7 @@ Content-Type: application/json
- `ADMIN_USERNAME`
- `ADMIN_PASSWORD`
- `SESSION_SECRET`
- `WEBHOOK_TOKEN`
- `DATABASE_PATH`
- `RETENTION_DAYS`
- `MAX_DELIVERY_ATTEMPTS`

2
app/config.py
View File

@ -13,6 +13,7 @@ class Settings:
admin_username: str = "admin"
admin_password: str = "change-me-now"
session_secret: str = "change-this-session-secret"
webhook_token: str = ""
retention_days: int = 30
max_delivery_attempts: int = 3
retry_backoff_seconds: int = 60
@ -27,6 +28,7 @@ def get_settings() -> Settings:
admin_username=os.getenv("ADMIN_USERNAME", "admin"),
admin_password=os.getenv("ADMIN_PASSWORD", "change-me-now"),
session_secret=os.getenv("SESSION_SECRET", "change-this-session-secret"),
webhook_token=os.getenv("WEBHOOK_TOKEN", ""),
retention_days=int(os.getenv("RETENTION_DAYS", "30")),
max_delivery_attempts=int(os.getenv("MAX_DELIVERY_ATTEMPTS", "3")),
retry_backoff_seconds=int(os.getenv("RETRY_BACKOFF_SECONDS", "60")),

63
app/dispatcher.py
View File

@ -14,7 +14,6 @@ from app.db import Database, from_json, now_iso, to_json
UTC = timezone.utc
REQUIRED_ALERT_FIELDS = ("timeframe", "symbol", "strategy")
TEMPLATE_PATTERN = re.compile(r"{{\s*([a-zA-Z0-9_.-]+)\s*}}|(?<!{){\s*([a-zA-Z0-9_.-]+)\s*}(?!})")
@ -23,13 +22,10 @@ class ValidationError(ValueError):
def normalize_alert(payload: dict[str, Any]) -> dict[str, Any]:
missing = [field for field in REQUIRED_ALERT_FIELDS if not str(payload.get(field, "")).strip()]
if missing:
raise ValidationError(f"Missing required fields: {', '.join(missing)}")
normalized = dict(payload)
normalized["timeframe"] = str(payload["timeframe"]).strip()
normalized["symbol"] = str(payload["symbol"]).strip().upper()
normalized["strategy"] = str(payload["strategy"]).strip()
normalized["timeframe"] = str(payload.get("timeframe", "")).strip()
normalized["symbol"] = str(payload.get("symbol", "")).strip().upper()
normalized["strategy"] = str(payload.get("strategy", "")).strip()
if "price" in normalized and normalized["price"] not in (None, ""):
try:
normalized["price"] = float(normalized["price"])
@ -59,9 +55,9 @@ def render_template(template: str, alert: dict[str, Any]) -> str:
def default_body(alert: dict[str, Any]) -> str:
action = alert.get("action") or alert.get("signal") or "alert"
lines = [
f"TradingView 信号: {alert['symbol']}",
f"周期: {alert['timeframe']}",
f"策略: {alert['strategy']}",
f"TradingView 信号: {alert.get('symbol') or '-'}",
f"周期: {alert.get('timeframe') or '-'}",
f"策略: {alert.get('strategy') or '-'}",
f"动作: {action}",
]
if alert.get("price") is not None:
@ -75,12 +71,9 @@ def build_feishu_message(alert: dict[str, Any], rule: dict[str, Any] | None = No
rule = rule or {}
title_template = rule.get("card_title_template") or "TradingView {{symbol}} {{action}}"
body_template = rule.get("card_body_template") or default_body(alert)
title = render_template(title_template, alert).strip() or f"TradingView {alert['symbol']}"
title = render_template(title_template, alert).strip() or f"TradingView {alert.get('symbol') or 'Alert'}"
body = render_template(body_template, alert).strip() or default_body(alert)
if rule.get("message_type") == "text":
return {"msg_type": "text", "content": {"text": f"{title}\n{body}"}}
return {
"msg_type": "interactive",
"card": {
@ -99,7 +92,7 @@ def build_feishu_message(alert: dict[str, Any], rule: dict[str, Any] | None = No
"elements": [
{
"tag": "plain_text",
"content": f"{alert['symbol']} · {alert['timeframe']} · {alert['strategy']}",
"content": f"{alert.get('symbol') or '-'} · {alert.get('timeframe') or '-'} · {alert.get('strategy') or '-'}",
}
],
},
@ -113,22 +106,35 @@ class Dispatcher:
self.db = db
self.settings = settings
def find_matching_rule(self, alert: dict[str, Any]) -> dict[str, Any] | None:
normalized = normalize_alert(alert)
with self.db.connect() as conn:
row = conn.execute(
"""
SELECT * FROM routing_rules
WHERE enabled = 1
AND (timeframe = '' OR timeframe = ?)
AND (symbol = '' OR upper(symbol) = ?)
AND (strategy = '' OR strategy = ?)
AND (timeframe <> '' OR symbol <> '' OR strategy <> '')
ORDER BY priority ASC,
(
CASE WHEN timeframe <> '' THEN 1 ELSE 0 END +
CASE WHEN symbol <> '' THEN 1 ELSE 0 END +
CASE WHEN strategy <> '' THEN 1 ELSE 0 END
) DESC,
id ASC
LIMIT 1
""",
(normalized["timeframe"], normalized["symbol"], normalized["strategy"]),
).fetchone()
return dict(row) if row else None
def receive_alert(self, payload: dict[str, Any]) -> dict[str, Any]:
alert = normalize_alert(payload)
created_at = now_iso()
with self.db.connect() as conn:
rule = conn.execute(
"""
SELECT * FROM routing_rules
WHERE enabled = 1
AND timeframe = ?
AND upper(symbol) = ?
AND strategy = ?
ORDER BY priority ASC, id ASC
LIMIT 1
""",
(alert["timeframe"], alert["symbol"], alert["strategy"]),
).fetchone()
rule = self.find_matching_rule(alert)
status = "matched" if rule else "unmatched"
cur = conn.execute(
@ -204,8 +210,7 @@ class Dispatcher:
with self.db.connect() as conn:
rows = conn.execute(
"""
SELECT d.*, a.payload
, r.message_type, r.card_title_template, r.card_body_template
SELECT d.*, a.payload, r.card_title_template, r.card_body_template
FROM deliveries d
JOIN alerts a ON a.id = d.alert_id
LEFT JOIN routing_rules r ON r.id = d.rule_id

329
app/server.py
View File

@ -4,6 +4,8 @@ import html
import json
import mimetypes
import os
import urllib.error
import urllib.request
from http import HTTPStatus
from http.cookies import SimpleCookie
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
@ -13,7 +15,7 @@ from urllib.parse import parse_qs, urlparse
from app.auth import COOKIE_NAME, check_credentials, hash_password, is_valid_session, make_session_cookie
from app.config import Settings, get_settings
from app.db import Database, from_json, now_iso, to_json
from app.dispatcher import Dispatcher, ValidationError
from app.dispatcher import Dispatcher, ValidationError, build_feishu_message, normalize_alert
class AppContext:
@ -103,8 +105,16 @@ class Handler(BaseHTTPRequestHandler):
self.render_dashboard()
elif parsed.path == "/targets":
self.render_targets()
elif parsed.path == "/targets/delete":
self.render_target_delete(parsed)
elif parsed.path == "/rules":
self.render_rules()
elif parsed.path == "/rules/new":
self.render_rule_new()
elif parsed.path == "/rules/edit":
self.render_rule_edit(parsed)
elif parsed.path == "/rules/delete":
self.render_rule_delete(parsed)
elif parsed.path == "/logs":
self.render_logs()
elif parsed.path == "/test":
@ -134,9 +144,11 @@ class Handler(BaseHTTPRequestHandler):
"/targets/create": self.create_target,
"/targets/update": self.update_target,
"/targets/delete": self.delete_target,
"/targets/test": self.test_target,
"/rules/create": self.create_rule,
"/rules/update": self.update_rule,
"/rules/delete": self.delete_rule,
"/rules/preview": self.preview_rule,
"/test/send": self.send_test,
"/account/password": self.change_password,
"/deliveries/retry": self.retry_deliveries,
@ -261,6 +273,12 @@ class Handler(BaseHTTPRequestHandler):
self.wfile.write(content)
def handle_tradingview_webhook(self) -> None:
if self.context.settings.webhook_token:
query = parse_qs(urlparse(self.path).query)
token = self.headers.get("X-Webhook-Token") or query.get("token", [""])[-1]
if token != self.context.settings.webhook_token:
json_response(self, 401, {"error": "Invalid webhook token"})
return
try:
payload = parse_json_body(self)
result = self.context.dispatcher.receive_alert(payload)
@ -290,6 +308,24 @@ class Handler(BaseHTTPRequestHandler):
return {"alerts": [dict(row) for row in alerts], "deliveries": [dict(row) for row in deliveries]}
def render_dashboard(self) -> None:
host = self.headers.get("Host", f"localhost:{self.context.settings.port}")
scheme = self.headers.get("X-Forwarded-Proto", "http")
base_url = f"{scheme}://{host}"
webhook_url = f"{base_url}/webhook/tradingview"
token = self.context.settings.webhook_token
webhook_url_with_token = f"{webhook_url}?token={token}" if token else webhook_url
token_block = (
f"""<div class="copy-row"><span>Webhook Token</span><code>{html.escape(token)}</code><button type="button" data-copy="{html.escape(token)}">复制</button></div>
<div class="copy-row"><span>Header 方式</span><code>X-Webhook-Token: {html.escape(token)}</code><button type="button" data-copy="X-Webhook-Token: {html.escape(token)}">复制</button></div>"""
if token
else """<p class="warning">当前未设置 WEBHOOK_TOKEN，任何知道地址的人都可以提交 alert。生产环境建议设置。</p>"""
)
webhook_panel = f"""<section class="panel">
<h2>TradingView Webhook 配置</h2>
<div class="copy-row"><span>Webhook URL</span><code>{html.escape(webhook_url_with_token)}</code><button type="button" data-copy="{html.escape(webhook_url_with_token)}">复制</button></div>
<div class="copy-row"><span>纯 URL</span><code>{html.escape(webhook_url)}</code><button type="button" data-copy="{html.escape(webhook_url)}">复制</button></div>
{token_block}
</section>"""
with self.context.db.connect() as conn:
counts = {
"alerts": conn.execute("SELECT COUNT(*) AS c FROM alerts").fetchone()["c"],
@ -308,7 +344,7 @@ class Handler(BaseHTTPRequestHandler):
f"<tr><td>{row['id']}</td><td>{html.escape(row['symbol'])}</td><td>{html.escape(row['timeframe'])}</td><td>{html.escape(row['strategy'])}</td><td><span class='status'>{html.escape(row['status'])}</span></td><td>{row['created_at']}</td></tr>"
for row in recent
)
self.send_html("概览", f"<header><h1>概览</h1><p>结构化 alert 分发、飞书转发和重试状态。</p></header><section class='metrics'>{cards}</section><section><h2>最近 Alert</h2><table><thead><tr><th>ID</th><th>品种</th><th>周期</th><th>策略</th><th>状态</th><th>时间</th></tr></thead><tbody>{rows}</tbody></table></section>")
self.send_html("概览", f"<header><h1>概览</h1><p>结构化 alert 分发、飞书转发和重试状态。</p></header>{webhook_panel}<section class='metrics'>{cards}</section><section><h2>最近 Alert</h2><table><thead><tr><th>ID</th><th>品种</th><th>周期</th><th>策略</th><th>状态</th><th>时间</th></tr></thead><tbody>{rows}</tbody></table></section>")
def render_targets(self) -> None:
targets = self.list_targets()
@ -319,7 +355,8 @@ class Handler(BaseHTTPRequestHandler):
<td class="url"><input form="target-update-{target['id']}" name="webhook_url" value="{html.escape(target['webhook_url'])}" type="url" required></td>
<td><label class="check"><input form="target-update-{target['id']}" name="enabled" type="checkbox" {'checked' if target['enabled'] else ''}> 启用</label></td>
<td><form id="target-update-{target['id']}" class="inline" method="post" action="/targets/update"></form><button form="target-update-{target['id']}" type="submit">更新</button>
<form class="inline" method="post" action="/targets/delete"><input type="hidden" name="id" value="{target['id']}"><button class="danger" type="submit">删除</button></form>
<form class="inline" method="post" action="/targets/test"><input type="hidden" name="id" value="{target['id']}"><button type="submit">测试</button></form>
<a class="button-link danger-link" href="/targets/delete?id={target['id']}">删除</a>
</td></tr>"""
for target in targets
)
@ -330,55 +367,201 @@ class Handler(BaseHTTPRequestHandler):
<label class="check"><input name="enabled" type="checkbox" checked> 启用</label>
<button type="submit">保存目标</button>
</form>"""
self.send_html("飞书 Webhook", f"<header><h1>飞书 Webhook</h1><p>维护所有可分发的飞书机器人地址。</p></header>{form}<table><thead><tr><th>ID</th><th>名称</th><th>URL</th><th>状态</th><th>操作</th></tr></thead><tbody>{rows}</tbody></table>")
notice = getattr(self, "_target_notice", "")
self.send_html("飞书 Webhook", f"<header><h1>飞书 Webhook</h1><p>维护所有可分发的飞书机器人地址。</p></header>{notice}{form}<table><thead><tr><th>ID</th><th>名称</th><th>URL</th><th>状态</th><th>操作</th></tr></thead><tbody>{rows}</tbody></table>")
def render_target_delete(self, parsed: Any) -> None:
target_id = parse_qs(parsed.query).get("id", [""])[-1]
with self.context.db.connect() as conn:
target = conn.execute("SELECT * FROM webhook_targets WHERE id = ?", (target_id,)).fetchone()
if not target:
self.send_error(404)
return
body = f"""<header><h1>删除飞书 Webhook</h1><p>请确认是否删除这个飞书目标。</p></header>
<section class="panel narrow">
<h2>{html.escape(target['name'])}</h2>
<p class="url">{html.escape(target['webhook_url'])}</p>
<form method="post" action="/targets/delete" class="actions">
<input type="hidden" name="id" value="{target['id']}">
<button class="danger" type="submit">确认删除</button>
<a class="button-link secondary" href="/targets">取消</a>
</form>
</section>"""
self.send_html("删除飞书 Webhook", body)
def render_rules(self) -> None:
targets = self.list_targets()
rules = self.list_rules()
target_names = {target["id"]: target["name"] for target in targets}
rows = ""
for rule in rules:
message_type_options = "".join(
f'<option value="{value}" {"selected" if rule["message_type"] == value else ""}>{label}</option>'
for value, label in [("card", "Card"), ("text", "Text")]
conditions = [
f"周期={html.escape(rule['timeframe'])}" if rule["timeframe"] else "",
f"品种={html.escape(rule['symbol'])}" if rule["symbol"] else "",
f"策略={html.escape(rule['strategy'])}" if rule["strategy"] else "",
]
target_name = target_names.get(rule["target_ids"][0], "-") if rule["target_ids"] else "-"
rows += f"""<tr>
<td>{rule['id']}</td>
<td>{html.escape(rule['name'])}</td>
<td>{'<br>'.join(item for item in conditions if item) or '-'}</td>
<td>{rule['priority']}</td>
<td>{html.escape(target_name)}</td>
<td><span class="status">{'启用' if rule['enabled'] else '停用'}</span></td>
<td><a class="button-link" href="/rules/edit?id={rule['id']}">编辑</a><a class="button-link danger-link" href="/rules/delete?id={rule['id']}">删除</a></td>
</tr>"""
header = """<header class="page-header"><div><h1>路由规则</h1><p>周期、品种、策略至少填写一个；空字段表示不限。消息统一用飞书卡片发送。</p></div><a class="button-link" href="/rules/new">新增规则</a></header>"""
self.send_html("路由规则", f"{header}<table><thead><tr><th>ID</th><th>名称</th><th>匹配条件</th><th>优先级</th><th>发送到</th><th>状态</th><th>操作</th></tr></thead><tbody>{rows}</tbody></table>")
def render_rule_form(
self,
title: str,
action: str,
rule: dict[str, Any] | None = None,
preview_html: str = "",
sample_payload: str | None = None,
) -> None:
targets = self.list_targets()
rule = rule or {
"id": "",
"name": "",
"timeframe": "",
"symbol": "",
"strategy": "",
"priority": 100,
"card_title_template": "TradingView {{symbol}} {{action}}",
"card_body_template": "**品种**: {{symbol}}\n**周期**: {{timeframe}}\n**策略**: {{strategy}}\n**动作**: {{action}}\n**价格**: {{price}}",
"target_ids": [],
"enabled": 1,
}
selected_targets = target_select_options(targets, rule.get("target_ids", []), placeholder=True)
hidden_id = f'<input type="hidden" name="id" value="{rule["id"]}">' if rule.get("id") else ""
button_text = "保存修改" if rule.get("id") else "创建规则"
sample_payload = sample_payload or json.dumps(
{"timeframe": "5m", "symbol": "BTCUSDT", "strategy": "breakout", "action": "buy", "price": 68000},
ensure_ascii=False,
indent=2,
)
selected_targets = target_select_options(targets, rule["target_ids"], placeholder=True)
rows += f"""<tr data-message-form>
<td>{rule['id']}<input form="rule-update-{rule['id']}" type="hidden" name="id" value="{rule['id']}"></td>
<td><input form="rule-update-{rule['id']}" name="name" value="{html.escape(rule['name'])}" required></td>
<td><input form="rule-update-{rule['id']}" name="timeframe" value="{html.escape(rule['timeframe'])}" required></td>
<td><input form="rule-update-{rule['id']}" name="symbol" value="{html.escape(rule['symbol'])}" required></td>
<td><input form="rule-update-{rule['id']}" name="strategy" value="{html.escape(rule['strategy'])}" required></td>
<td><input form="rule-update-{rule['id']}" name="priority" type="number" value="{rule['priority']}" required></td>
<td><select class="select-compact" form="rule-update-{rule['id']}" name="message_type" data-message-type>{message_type_options}</select></td>
<td><textarea form="rule-update-{rule['id']}" name="card_title_template" rows="2" data-title-template>{html.escape(rule['card_title_template'])}</textarea></td>
<td><textarea form="rule-update-{rule['id']}" name="card_body_template" rows="4" data-body-template>{html.escape(rule['card_body_template'])}</textarea></td>
<td><select class="select-target" form="rule-update-{rule['id']}" name="target_ids" required>{selected_targets}</select></td>
<td><label class="check"><input form="rule-update-{rule['id']}" name="enabled" type="checkbox" {'checked' if rule['enabled'] else ''}> 启用</label></td>
<td><form id="rule-update-{rule['id']}" class="inline" method="post" action="/rules/update"></form><button form="rule-update-{rule['id']}" type="submit">更新</button><form class="inline" method="post" action="/rules/delete"><input type="hidden" name="id" value="{rule['id']}"><button class="danger" type="submit">删除</button></form></td></tr>"""
create_target_options = target_select_options(targets, placeholder=True)
form = f"""<form class="panel" method="post" action="/rules/create" data-message-form>
<h2>新增路由规则</h2>
body = f"""<header><h1>{html.escape(title)}</h1><p>消息统一使用飞书卡片。周期、品种、策略至少填写一个，空字段表示不限。</p></header>
<form class="panel rule-form" method="post" action="{action}">
{hidden_id}
<div class="grid">
<label>规则名<input name="name" required></label>
<label>周期<input name="timeframe" placeholder="5m" required></label>
<label>品种<input name="symbol" placeholder="BTCUSDT" required></label>
<label>策略<input name="strategy" placeholder="breakout" required></label>
<label>优先级<input name="priority" type="number" value="100" required></label>
<label>规则名<input name="name" value="{html.escape(str(rule['name']))}" required></label>
<label>周期<input name="timeframe" value="{html.escape(str(rule['timeframe']))}" placeholder="5m，空=不限"></label>
<label>品种<input name="symbol" value="{html.escape(str(rule['symbol']))}" placeholder="BTCUSDT，空=不限"></label>
<label>策略<input name="strategy" value="{html.escape(str(rule['strategy']))}" placeholder="breakout，空=不限"></label>
<label>优先级<input name="priority" type="number" value="{rule['priority']}" required></label>
</div>
<div>
<label class="field-compact">消息类型<select class="select-compact" name="message_type" data-message-type><option value="card" selected>Card</option><option value="text">Text</option></select></label>
<label><span data-title-label>卡片标题模板</span><input name="card_title_template" value="TradingView {{{{symbol}}}} {{{{action}}}}" data-title-template required></label>
<label><span data-body-label>卡片正文模板</span><textarea name="card_body_template" rows="5" data-body-template>**品种**: {{{{symbol}}}}
**周期**: {{{{timeframe}}}}
**策略**: {{{{strategy}}}}
**动作**: {{{{action}}}}
**价格**: {{{{price}}}}</textarea></label>
<label class="field-target">发送到<select class="select-target" name="target_ids" required>{create_target_options}</select></label>
<label>卡片标题模板<input name="card_title_template" value="{html.escape(str(rule['card_title_template']))}" required></label>
<label>卡片正文模板<textarea name="card_body_template" rows="6">{html.escape(str(rule['card_body_template']))}</textarea></label>
<label class="field-target">发送到<select class="select-target" name="target_ids" required>{selected_targets}</select></label>
<label class="check"><input name="enabled" type="checkbox" {'checked' if rule.get('enabled') else ''}> 启用</label>
<h2>规则命中与卡片预览</h2>
<input type="hidden" name="source_action" value="{html.escape(action)}">
<label>样例 Alert JSON<textarea name="sample_payload" rows="9">{html.escape(sample_payload)}</textarea></label>
<div class="actions"><button type="submit">{button_text}</button><button type="submit" formaction="/rules/preview">预览命中和卡片</button><a class="button-link secondary" href="/rules">返回列表</a></div>
</form>{preview_html}"""
self.send_html(title, body)
def render_rule_new(self) -> None:
self.render_rule_form("新增路由规则", "/rules/create")
def render_rule_edit(self, parsed: Any) -> None:
query = parse_qs(parsed.query)
rule_id = query.get("id", [""])[-1]
with self.context.db.connect() as conn:
rule = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
if not rule:
self.send_error(404)
return
rule_dict = dict(rule)
rule_dict["target_ids"] = from_json(rule_dict["target_ids"], [])
self.render_rule_form("编辑路由规则", "/rules/update", rule_dict)
def render_rule_delete(self, parsed: Any) -> None:
rule_id = parse_qs(parsed.query).get("id", [""])[-1]
with self.context.db.connect() as conn:
rule = conn.execute("SELECT * FROM routing_rules WHERE id = ?", (rule_id,)).fetchone()
if not rule:
self.send_error(404)
return
body = f"""<header><h1>删除路由规则</h1><p>请确认是否删除这条规则。</p></header>
<section class="panel narrow">
<h2>{html.escape(rule['name'])}</h2>
<p>删除后不会再匹配对应 alert，已有日志不受影响。</p>
<form method="post" action="/rules/delete" class="actions">
<input type="hidden" name="id" value="{rule['id']}">
<button class="danger" type="submit">确认删除</button>
<a class="button-link secondary" href="/rules">取消</a>
</form>
</section>"""
self.send_html("删除路由规则", body)
def build_rule_from_form(self, form: dict[str, list[str]]) -> dict[str, Any]:
return {
"id": form.get("id", [""])[-1],
"name": form.get("name", [""])[-1].strip(),
"timeframe": form.get("timeframe", [""])[-1].strip(),
"symbol": form.get("symbol", [""])[-1].strip().upper(),
"strategy": form.get("strategy", [""])[-1].strip(),
"priority": int(form.get("priority", ["100"])[-1] or 100),
"card_title_template": form.get("card_title_template", ["TradingView {{symbol}} {{action}}"])[-1].strip(),
"card_body_template": form.get("card_body_template", [""])[-1].strip(),
"target_ids": [int(value) for value in form.get("target_ids", []) if value],
"enabled": 1 if form.get("enabled", [""])[-1] == "on" else 0,
}
def preview_rule(self) -> None:
form = parse_form_multi(self)
rule = self.build_rule_from_form(form)
sample_payload = form.get("sample_payload", ["{}"])[-1]
source_action = form.get("source_action", ["/rules/create"])[-1]
title = "编辑路由规则" if rule.get("id") else "新增路由规则"
try:
alert = normalize_alert(json.loads(sample_payload))
message = build_feishu_message(alert, rule)
matched = self.context.dispatcher.find_matching_rule(alert)
current_matches = self.rule_matches_alert(rule, alert)
preview_html = self.render_preview_result(rule, message, matched, current_matches)
except (json.JSONDecodeError, ValidationError, ValueError) as exc:
preview_html = f"""<section class="result-panel error"><h2>预览失败</h2><p>{html.escape(str(exc))}</p></section>"""
self.render_rule_form(title, source_action, rule, preview_html, sample_payload)
def rule_matches_alert(self, rule: dict[str, Any], alert: dict[str, Any]) -> bool:
if not any((rule.get("timeframe"), rule.get("symbol"), rule.get("strategy"))):
return False
if rule.get("timeframe") and rule["timeframe"] != alert.get("timeframe"):
return False
if rule.get("symbol") and rule["symbol"].upper() != alert.get("symbol"):
return False
if rule.get("strategy") and rule["strategy"] != alert.get("strategy"):
return False
return True
def render_preview_result(
self,
rule: dict[str, Any],
message: dict[str, Any],
matched: dict[str, Any] | None,
current_matches: bool,
) -> str:
title = message["card"]["header"]["title"]["content"]
content = message["card"]["elements"][0]["text"]["content"]
matched_text = f"当前已保存规则 #{matched['id']} {matched['name']}" if matched else "没有已保存规则会命中"
current_text = "当前表单会匹配样例 Alert" if current_matches else "当前表单不会匹配样例 Alert"
return f"""<section class="result-panel success">
<h2>预览结果</h2>
<div class="result-grid">
<div><span>当前表单</span><strong>{html.escape(current_text)}</strong></div>
<div><span>系统实际命中</span><strong>{html.escape(matched_text)}</strong></div>
<div><span>规则优先级</span><strong>{rule.get('priority')}</strong></div>
<div><span>消息类型</span><strong>飞书卡片</strong></div>
</div>
<label class="check"><input name="enabled" type="checkbox" checked> 启用</label>
<button type="submit">保存规则</button>
</form>"""
self.send_html("路由规则", f"<header><h1>路由规则</h1><p>每条规则选择一个飞书 Webhook。模板支持 TradingView JSON 字段，例如 {{{{symbol}}}}、{{{{timeframe}}}}、{{{{strategy}}}}、{{{{price}}}}，嵌套字段可写 {{{{order.id}}}}。</p></header>{form}<table><thead><tr><th>ID</th><th>名称</th><th>周期</th><th>品种</th><th>策略</th><th>优先级</th><th>消息</th><th>标题模板</th><th>内容模板</th><th>发送到</th><th>状态</th><th>操作</th></tr></thead><tbody>{rows}</tbody></table>")
<div class="feishu-preview">
<div class="feishu-preview-header">{html.escape(title)}</div>
<pre>{html.escape(content)}</pre>
</div>
</section>"""
def render_logs(self) -> None:
logs = self.list_logs()
@ -444,9 +627,49 @@ class Handler(BaseHTTPRequestHandler):
conn.execute("DELETE FROM webhook_targets WHERE id = ?", (form["id"],))
redirect(self, "/targets")
def test_target(self) -> None:
form = parse_form(self)
with self.context.db.connect() as conn:
target = conn.execute("SELECT * FROM webhook_targets WHERE id = ?", (form.get("id"),)).fetchone()
if not target:
self.send_error(404)
return
message = build_feishu_message(
{"symbol": "TEST", "timeframe": "5m", "strategy": "webhook-test", "action": "test"},
{
"card_title_template": "TV Dispatch 测试消息",
"card_body_template": "**目标**: {{symbol}}\n**动作**: {{action}}\n这是一条飞书 Webhook 连通性测试。",
},
)
try:
data = json.dumps(message, ensure_ascii=False).encode()
request = urllib.request.Request(
target["webhook_url"],
data=data,
headers={"Content-Type": "application/json"},
method="POST",
)
with urllib.request.urlopen(request, timeout=self.context.settings.feishu_timeout_seconds) as response:
status = response.getcode()
self.render_targets_with_notice(f"测试消息已发送到 {html.escape(target['name'])}，HTTP {status}", success=True)
except urllib.error.HTTPError as exc:
self.render_targets_with_notice(f"测试发送失败：HTTP {exc.code}", success=False)
except Exception as exc:
self.render_targets_with_notice(f"测试发送失败：{html.escape(str(exc))}", success=False)
def render_targets_with_notice(self, message: str, success: bool) -> None:
self._target_notice = f"""<section class="result-panel {'success' if success else 'error'}"><h2>Webhook 测试</h2><p>{message}</p></section>"""
self.render_targets()
def create_rule(self) -> None:
form = parse_form_multi(self)
target_ids = [int(value) for value in form.get("target_ids", [])]
timeframe = form.get("timeframe", [""])[-1].strip()
symbol = form.get("symbol", [""])[-1].strip().upper()
strategy = form.get("strategy", [""])[-1].strip()
if not any((timeframe, symbol, strategy)):
self.send_error(400, "周期、品种、策略至少填写一个")
return
now = now_iso()
with self.context.db.connect() as conn:
conn.execute(
@ -460,11 +683,11 @@ class Handler(BaseHTTPRequestHandler):
""",
(
form.get("name", [""])[-1].strip(),
form.get("timeframe", [""])[-1].strip(),
form.get("symbol", [""])[-1].strip().upper(),
form.get("strategy", [""])[-1].strip(),
timeframe,
symbol,
strategy,
int(form.get("priority", ["100"])[-1]),
form.get("message_type", ["card"])[-1],
"card",
form.get("card_title_template", ["TradingView {{symbol}} {{action}}"])[-1].strip(),
form.get("card_body_template", [""])[-1].strip(),
1 if form.get("enabled", [""])[-1] == "on" else 0,
@ -484,6 +707,12 @@ class Handler(BaseHTTPRequestHandler):
def update_rule(self) -> None:
form = parse_form_multi(self)
target_ids = [int(value) for value in form.get("target_ids", [])]
timeframe = form.get("timeframe", [""])[-1].strip()
symbol = form.get("symbol", [""])[-1].strip().upper()
strategy = form.get("strategy", [""])[-1].strip()
if not any((timeframe, symbol, strategy)):
self.send_error(400, "周期、品种、策略至少填写一个")
return
with self.context.db.connect() as conn:
conn.execute(
"""
@ -495,11 +724,11 @@ class Handler(BaseHTTPRequestHandler):
""",
(
form.get("name", [""])[-1].strip(),
form.get("timeframe", [""])[-1].strip(),
form.get("symbol", [""])[-1].strip().upper(),
form.get("strategy", [""])[-1].strip(),
timeframe,
symbol,
strategy,
int(form.get("priority", ["100"])[-1]),
form.get("message_type", ["card"])[-1],
"card",
form.get("card_title_template", ["TradingView {{symbol}} {{action}}"])[-1].strip(),
form.get("card_body_template", [""])[-1].strip(),
1 if form.get("enabled", [""])[-1] == "on" else 0,

40
app/static/app.js
View File

@ -1,29 +1,15 @@
function updateMessageForm(scope) {
const typeSelect = scope.querySelector("[data-message-type]");
if (!typeSelect) return;
const isText = typeSelect.value === "text";
const titleLabel = scope.querySelector("[data-title-label]");
const bodyLabel = scope.querySelector("[data-body-label]");
const titleTemplate = scope.querySelector("[data-title-template]");
const bodyTemplate = scope.querySelector("[data-body-template]");
scope.classList.toggle("text-message", isText);
if (titleLabel) titleLabel.textContent = isText ? "文本标题模板" : "卡片标题模板";
if (bodyLabel) bodyLabel.textContent = isText ? "文本内容模板" : "卡片正文模板";
if (titleTemplate) {
titleTemplate.placeholder = isText ? "例如：TradingView {{symbol}}" : "例如：TradingView {{symbol}} {{action}}";
}
if (bodyTemplate) {
bodyTemplate.placeholder = isText ? "{{symbol}} {{timeframe}} {{strategy}} {{action}}" : "**品种**: {{symbol}}";
}
}
document.addEventListener("DOMContentLoaded", () => {
document.querySelectorAll("[data-message-form]").forEach((scope) => {
updateMessageForm(scope);
const typeSelect = scope.querySelector("[data-message-type]");
if (typeSelect) {
typeSelect.addEventListener("change", () => updateMessageForm(scope));
document.addEventListener("click", async (event) => {
const button = event.target.closest("[data-copy]");
if (!button) return;
const value = button.getAttribute("data-copy") || "";
try {
await navigator.clipboard.writeText(value);
const original = button.textContent;
button.textContent = "已复制";
setTimeout(() => {
button.textContent = original;
}, 1200);
} catch {
window.prompt("复制下面的内容", value);
}
});
});

106
app/static/styles.css
View File

@ -77,6 +77,13 @@ header {
margin-bottom: 24px;
}
.page-header {
display: flex;
align-items: flex-start;
justify-content: space-between;
gap: 18px;
}
h1,
h2 {
margin: 0 0 8px;
@ -181,33 +188,15 @@ td textarea {
max-width: 560px;
}
.field-compact,
.field-target {
width: fit-content;
max-width: 100%;
}
.select-compact {
width: 140px;
}
.select-target {
width: clamp(220px, 34vw, 360px);
}
td .select-compact {
min-width: 110px;
}
td .select-target {
min-width: 180px;
max-width: 260px;
}
.text-message [data-title-template] {
background: #f7f3e8;
}
.check {
display: inline-flex;
align-items: center;
@ -247,6 +236,57 @@ button:hover {
background: var(--danger);
}
.button-link {
display: inline-block;
border-radius: 6px;
padding: 10px 14px;
background: var(--accent);
color: white;
font: 800 14px ui-sans-serif, system-ui, sans-serif;
text-decoration: none;
margin-right: 8px;
}
.button-link.secondary {
background: #ece7da;
color: var(--ink);
}
.actions {
display: flex;
align-items: center;
gap: 10px;
margin-top: 8px;
}
.copy-row {
display: grid;
grid-template-columns: 150px minmax(0, 1fr) auto;
gap: 12px;
align-items: center;
margin-top: 12px;
font-family: ui-sans-serif, system-ui, sans-serif;
}
.copy-row span {
color: var(--muted);
font-weight: 700;
}
.copy-row code {
display: block;
overflow-wrap: anywhere;
border: 1px solid var(--line);
border-radius: 6px;
padding: 10px 12px;
background: #fff;
}
.warning {
margin-top: 12px;
color: var(--danger);
}
.result-panel {
border-radius: 8px;
border: 1px solid var(--line);
@ -264,6 +304,28 @@ button:hover {
border-left: 5px solid var(--danger);
}
.feishu-preview {
border: 1px solid var(--line);
border-radius: 8px;
overflow: hidden;
background: #fff;
margin-top: 16px;
}
.feishu-preview-header {
background: #1d4ed8;
color: #fff;
padding: 12px 14px;
font: 800 15px ui-sans-serif, system-ui, sans-serif;
}
.feishu-preview pre {
margin: 0;
background: #fff;
color: var(--ink);
border-radius: 0;
}
.result-grid {
display: grid;
grid-template-columns: repeat(4, minmax(0, 1fr));
@ -372,6 +434,10 @@ th {
width: auto;
}
.page-header {
display: grid;
}
.shell {
margin-left: 0;
padding: 22px;
@ -382,4 +448,8 @@ th {
.result-grid {
grid-template-columns: 1fr;
}
.copy-row {
grid-template-columns: 1fr;
}
}

4
docker-compose.yml
View File

@ -2,11 +2,12 @@ services:
dispatcher:
build: .
ports:
- "8000:8000"
- "8030:8000"
environment:
ADMIN_USERNAME: admin
ADMIN_PASSWORD: change-me-now
SESSION_SECRET: replace-with-a-long-random-secret
WEBHOOK_TOKEN: replace-with-a-shared-webhook-secret
RETENTION_DAYS: 30
MAX_DELIVERY_ATTEMPTS: 3
RETRY_BACKOFF_SECONDS: 60
@ -20,6 +21,7 @@ services:
ADMIN_USERNAME: admin
ADMIN_PASSWORD: change-me-now
SESSION_SECRET: replace-with-a-long-random-secret
WEBHOOK_TOKEN: replace-with-a-shared-webhook-secret
RETENTION_DAYS: 30
MAX_DELIVERY_ATTEMPTS: 3
RETRY_BACKOFF_SECONDS: 60

53
tests/test_dispatcher.py
View File

@ -34,7 +34,15 @@ class DispatcherTest(unittest.TestCase):
)
return int(cur.lastrowid)
def add_rule(self, target_id: int, priority: int = 100, name: str = "rule") -> int:
def add_rule(
self,
target_id: int,
priority: int = 100,
name: str = "rule",
timeframe: str = "5m",
symbol: str = "BTCUSDT",
strategy: str = "breakout",
) -> int:
now = now_iso()
with self.db.connect() as conn:
cur = conn.execute(
@ -45,16 +53,12 @@ class DispatcherTest(unittest.TestCase):
card_title_template, card_body_template, enabled, target_ids,
created_at, updated_at
)
VALUES (?, '5m', 'BTCUSDT', 'breakout', ?, 'card', 'Signal {{symbol}}', 'Price {{price}}', 1, ?, ?, ?)
VALUES (?, ?, ?, ?, ?, 'card', 'Signal {{symbol}}', 'Price {{price}}', 1, ?, ?, ?)
""",
(name, priority, to_json([target_id]), now, now),
(name, timeframe, symbol, strategy, priority, to_json([target_id]), now, now),
)
return int(cur.lastrowid)
def test_missing_required_fields_are_rejected(self) -> None:
with self.assertRaises(ValidationError):
self.dispatcher.receive_alert({"symbol": "BTCUSDT"})
def test_unmatched_alert_is_stored(self) -> None:
result = self.dispatcher.receive_alert({"timeframe": "15m", "symbol": "ETHUSDT", "strategy": "trend"})
self.assertEqual(result["status"], "unmatched")
@ -78,6 +82,35 @@ class DispatcherTest(unittest.TestCase):
delivery = conn.execute("SELECT * FROM deliveries WHERE alert_id = ?", (result["alert_id"],)).fetchone()
self.assertEqual(delivery["target_id"], fast_target)
def test_single_dimension_rule_matches(self) -> None:
target_id = self.add_target()
rule_id = self.add_rule(target_id, timeframe="", symbol="BTCUSDT", strategy="")
result = self.dispatcher.receive_alert({"symbol": "btcusdt", "action": "buy"})
self.assertEqual(result["status"], "matched")
self.assertEqual(result["matched_rule_id"], rule_id)
def test_more_specific_rule_wins_when_priority_ties(self) -> None:
broad_target = self.add_target("broad")
specific_target = self.add_target("specific")
broad_rule = self.add_rule(broad_target, priority=10, name="broad", timeframe="", symbol="BTCUSDT", strategy="")
specific_rule = self.add_rule(specific_target, priority=10, name="specific")
result = self.dispatcher.receive_alert({"timeframe": "5m", "symbol": "BTCUSDT", "strategy": "breakout"})
self.assertEqual(result["matched_rule_id"], specific_rule)
self.assertNotEqual(result["matched_rule_id"], broad_rule)
def test_find_matching_rule_preview(self) -> None:
target_id = self.add_target()
rule_id = self.add_rule(target_id, timeframe="", symbol="BTCUSDT", strategy="")
rule = self.dispatcher.find_matching_rule({"symbol": "btcusdt"})
self.assertIsNotNone(rule)
self.assertEqual(rule["id"], rule_id)
def test_failed_delivery_is_marked_for_retry(self) -> None:
target_id = self.add_target()
self.add_rule(target_id)
@ -106,7 +139,7 @@ class DispatcherTest(unittest.TestCase):
self.assertEqual(message["card"]["header"]["title"]["content"], "BTCUSDT buy")
self.assertEqual(message["card"]["elements"][0]["text"]["content"], "**价格** 68000")
def test_template_accepts_legacy_single_braces(self) -> None:
def test_template_accepts_legacy_single_braces_and_still_sends_card(self) -> None:
message = build_feishu_message(
{"timeframe": "5m", "symbol": "BTCUSDT", "strategy": "breakout", "action": "buy", "price": 68000},
{
@ -116,7 +149,9 @@ class DispatcherTest(unittest.TestCase):
},
)
self.assertEqual(message["content"]["text"], "TradingView BTCUSDT buy\n价格 68000")
self.assertEqual(message["msg_type"], "interactive")
self.assertEqual(message["card"]["header"]["title"]["content"], "TradingView BTCUSDT buy")
self.assertEqual(message["card"]["elements"][0]["text"]["content"], "价格 68000")
if __name__ == "__main__":