v4.5.38: 实时数据查询硬规则(回复前必查)+ check_account.py封装
This commit is contained in:
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
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()
|
||||
Reference in New Issue
Block a user