81 lines
2.3 KiB
Python
81 lines
2.3 KiB
Python
"""
|
|
v4.5.24 sanitize_reply — 把 agent 写完的回复做最后一道 grep 拦截。
|
|
|
|
背景: v4.5.21~23 三次写"零字符沉默"规则,但 agent 在 250+ 连发 SKHYNIX 加仓
|
|
场景中**实战违反 ~290 次**(两次事故,见 references/v4.5.24-*-violation.md)。
|
|
|
|
根因: 文档规则 agent 不主动遵守。修复必须**代码层面拦截**。
|
|
|
|
用法:
|
|
reply_text = build_reply(signal, position_state)
|
|
reply_text = sanitize_reply(reply_text)
|
|
if reply_text:
|
|
send_telegram(reply_text) # 只在非空时发
|
|
|
|
CLI 用法 (调试):
|
|
echo "SKHYNIX long 0.216张@5x 浮盈+\$5" | python3 sanitize_reply.py
|
|
# 输出: (空)
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
|
|
|
|
# 黑名单: 命中这些 token → 视为持仓状态泄漏 / 重复信号推送
|
|
# 实测覆盖 v4.5.24 两次事故的所有违规 pattern
|
|
BANNED_PATTERNS = re.compile(
|
|
r'浮盈|仍持仓|重复.*已推|无execute|⏭️|'
|
|
r'SKHYNIX long|ETH short|MU short|GRAM short|SKHY short|'
|
|
r'CL short|SPCX long|SPCX short|BTC long|ETH long|'
|
|
r'已跟|浮亏|强平价|已平仓|执行成功|信号已推|'
|
|
r'未execute|浮盈\+|浮亏\+|USDT \$[0-9]|'
|
|
r'avgPx|markPx|lever|liqPx|imr|notionalUsd|'
|
|
r'开仓价|当前价|保证金:|\$ [0-9]|ct_val'
|
|
)
|
|
|
|
# 例外直通: 这些 token 出现时不视为违规(用于首次 execute / 平仓触发 / 主动询问)
|
|
EXCEPTION_TOKENS = (
|
|
'执行成功', # execute 首次成功上报
|
|
'已平仓', # 平仓触发后回报
|
|
)
|
|
|
|
|
|
def sanitize_reply(text: str) -> str:
|
|
"""对 agent 写完的回复做硬阻断。
|
|
|
|
Args:
|
|
text: agent 写完的回复文本
|
|
|
|
Returns:
|
|
合规的文本(可能为空字符串)
|
|
"""
|
|
if not text:
|
|
return text
|
|
|
|
# 例外直通: 含 EXCEPTION_TOKENS 任一 → 视为合规
|
|
for token in EXCEPTION_TOKENS:
|
|
if token in text:
|
|
return text
|
|
|
|
# 黑名单阻断
|
|
if BANNED_PATTERNS.search(text):
|
|
return ""
|
|
|
|
return text
|
|
|
|
|
|
def main():
|
|
"""CLI 入口: 从 stdin 读文本, 输出 sanitize 后结果"""
|
|
if len(sys.argv) > 1:
|
|
text = " ".join(sys.argv[1:])
|
|
else:
|
|
text = sys.stdin.read().strip()
|
|
|
|
sanitized = sanitize_reply(text)
|
|
if sanitized:
|
|
print(sanitized)
|
|
# else: 静默(零字符)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |