AI 自进化LOOP ENGINEER交易系统
2
Follow
487
Followers
Source
Python
# -*- coding: utf-8 -*-
"""
单品种 AI 自进化交易系统 - 发明者量化(FMZ.COM) Python V1
适用标的：BTC / 黄金 / 原油 等任意单一 TradFi 或加密货币永续合约
=====================================================================
设计核心：Decide -> Execute -> Reflect -> Evolve 闭环自进化
=====================================================================
1. 指标层：纯 Python 手写多组技术指标(趋势/动量/波动率/量能)，不依赖
平台 TA 库具体版本签名，保证跨环境稳定运行。
2. 新闻层：BraveSearch 抓取该品种相关新闻，去重缓存进数据库。
3. 决策层：LLM 综合"指标快照 + 新闻 + 当前 Playbook(经验库)"输出结构化
交易信号，要求技术面与消息面不冲突才允许开仓/加仓。
4. 执行层：仓位/止损/移动止盈参数从 Playbook 读取(不是写死常量)，
支持同向加仓、反向平仓再反手，内置硬止损与移动止盈。
5. 复盘层：每笔平仓后，LLM 分析这笔交易的输赢原因，提炼中文经验写入
数据库(reflections 表)。
6. 进化层：每积累 N 笔复盘(或每隔固定时间)，LLM 读取近期复盘 + 整体
胜率统计，输出新版本 Playbook(新的 do/avoid 规则 + 参数建议)。
参数建议会被裁剪进人工设定的安全上下限，写回数据库后立即生效，
下一次决策自动使用最新 Playbook —— 这就是"自进化"的闭环。
安全设计：
- AI 不能修改代码，只能在人工设定的硬性范围内微调数值参数。
- 硬止损、单标的仓位上限、最大加仓次数是代码层强约束，AI 建议无法突破。
- 开仓强制要求"技术面 + 消息面"双重不冲突确认。
- 默认 notify 模式，仅通知不实盘。
=====================================================================
以下常量由平台参数注入(在 FMZ 策略"参数"面板中创建，无需在代码中硬编码)：
LLM_API_KEY / LLM_BASE_URL / LLM_MODEL LLM 接口配置
BRAVE_KEY BraveSearch Key(留空跳过新闻)
TRADE_MODE notify 仅通知 / trade 实盘
INSTRUMENT_KEYWORD 合约匹配关键字，如 "BTC_USDT" "XAU_USDT" "WTI_USDT"
INSTRUMENT_DISPLAY_NAME 中文显示名，如 "比特币" "黄金" "WTI原油"
NEWS_QUERY_TERMS 逗号分隔的新闻检索关键词，如 "美联储,ETF,减半"
KLINE_PERIOD_LABEL M15 / M30 / H1 / H4 / D1
CHECK_INTERVAL_S 主循环间隔秒
NEWS_REFRESH_S 新闻刷新间隔秒
DECISION_COOLDOWN_S 两次AI决策最小间隔秒
KLINE_LIMIT 拉取K线根数
MIN_CONFIDENCE / MIN_CONFIDENCE_FLOOR / MIN_CONFIDENCE_CEIL 最小置信度(初始值/进化下限/进化上限)
BASE_POS_PCT / BASE_POS_PCT_FLOOR / BASE_POS_PCT_CEIL 基础仓位比例(初始值/进化下限/进化上限)
MAX_SINGLE_POS_PCT 单标的最大仓位比例(硬约束，不参与进化)
MAX_TOTAL_ADD_COUNT 最大加仓次数(硬约束)
LEVERAGE 杠杆(硬约束)
HARD_STOP_PCT / HARD_STOP_PCT_MIN / HARD_STOP_PCT_MAX 硬止损%(初始值/进化下限/进化上限)
TRAIL_ACTIVATE_PCT / TRAIL_ACTIVATE_PCT_MIN / TRAIL_ACTIVATE_PCT_MAX 移动止盈激活%
TRAIL_GIVEBACK_PCT / TRAIL_GIVEBACK_PCT_MIN / TRAIL_GIVEBACK_PCT_MAX 移动止盈回撤%
EVOLUTION_TRADE_INTERVAL 每积累几笔已复盘交易触发一次进化
EVOLUTION_TIME_INTERVAL_S 兜底：每隔多久强制检查一次进化
EVOLUTION_MIN_TRADES 至少要有几笔复盘记录才允许第一次进化
REFLECTION_LOOKBACK 进化时回顾最近几笔复盘记录
=====================================================================
"""
import json
import math
import re
import time
import urllib.parse
import urllib.request
STORE_KEY = "ai_single_evolve_store_v1"
POSITION_STATE_KEY = "ai_single_evolve_pos_v1"
# =========================
# 主循环
# =========================
def main():
Log("单品种AI自进化交易系统启动 | {} | mode:{}".format(INSTRUMENT_DISPLAY_NAME, TRADE_MODE), "#00AAFF")
db_init()
store = load_store()
if not store.get("market"):
resolve_market(store)
load_or_init_playbook(store)
last_dashboard = 0
while True:
try:
handle_command(store)
if not store.get("market"):
resolve_market(store)
now = int(time.time())
if store.get("market"):
if BRAVE_KEY and now - store.get("last_news_refresh", 0) > NEWS_REFRESH_S:
news = fetch_instrument_news(store)
store["news_cache_recent"] = news
store["last_news_refresh"] = now
save_store(store)
if now - store.get("last_decision_time", 0) > DECISION_COOLDOWN_S:
run_decision_cycle(store)
store["last_decision_time"] = now
save_store(store)
monitor_position(store)
maybe_run_evolution(store)
if now - last_dashboard > 60:
show_dashboard(store)
last_dashboard = now
save_store(store)
except Exception as e:
Log("主循环异常: " + str(e), "#FF0000")
Sleep(CHECK_INTERVAL_S * 1000)
# =========================
# Store (工作记忆，轻量易变状态)
# =========================
def new_store():
return {
"market": None,
"indicator_snapshot": {},
"news_cache_recent": [],
"current_playbook": {},
"signals": [],
"step_logs": [],
"closed_trade_count_since_evolution": 0,
"evolution_paused": False,
"last_news_refresh": 0,
"last_decision_time": 0,
"last_evolution_check": 0,
}
def load_store():
store = _G(STORE_KEY)
if not store:
return new_store()
fresh = new_store()
for k, v in fresh.items():
if k not in store:
store[k] = v
return store
def save_store(store):
_G(STORE_KEY, store)
def record_step(store, step, data, color="#999999"):
row = {"time": int(time.time()), "step": step, "data": data}
store.setdefault("step_logs", []).insert(0, row)
store["step_logs"] = store["step_logs"][:300]
Log("STEP | {} | {}".format(step, json.dumps(data, ensure_ascii=False)[:400]), color)
def handle_command(store):
cmd = GetCommand()
if not cmd:
return
if cmd == "decision:force":
store["last_decision_time"] = 0
save_store(store)
Log("已强制触发下一轮决策", "#00CC66")
elif cmd == "evolution:trigger":
run_evolution(store, forced=True)
elif cmd == "evolution:pause":
store["evolution_paused"] = True
save_store(store)
Log("进化已暂停", "#FFAA00")
elif cmd == "evolution:resume":
store["evolution_paused"] = False
save_store(store)
Log("进化已恢复", "#00CC66")
elif cmd == "clear:store":
market = store.get("market")
store.clear()
store.update(new_store())
store["market"] = market
load_or_init_playbook(store)
save_store(store)
Log("运行态已重置(数据库历史交易/经验保留)", "#FFAA00")
elif cmd == "db:wipe":
for tbl in ["indicator_snapshot", "decisions", "trades", "reflections", "playbook", "news_cache"]:
db_exec("DELETE FROM {}".format(tbl))
store.clear()
store.update(new_store())
resolve_market(store)
load_or_init_playbook(store)
save_store(store)
Log("数据库与运行态已全部清空", "#FF0000")
# =========================
# 数据库层 (DBExec / SQLite - 长期记忆)
# =========================
def db_init():
db_exec('CREATE TABLE IF NOT EXISTS indicator_snapshot ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER NOT NULL, '
'price REAL, snapshot_json TEXT NOT NULL);')
db_exec('CREATE TABLE IF NOT EXISTS decisions ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER NOT NULL, '
'playbook_version INTEGER, action TEXT, direction TEXT, confidence INTEGER, '
'technical_view TEXT, news_view TEXT, reason TEXT, '
'indicators_json TEXT, news_json TEXT, executed INTEGER DEFAULT 0);')
db_exec('CREATE TABLE IF NOT EXISTS trades ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, open_time INTEGER, close_time INTEGER, '
'direction TEXT, entry_price REAL, exit_price REAL, qty REAL, pnl_pct REAL, '
'close_reason TEXT, decision_id INTEGER, reflected INTEGER DEFAULT 0);')
db_exec('CREATE TABLE IF NOT EXISTS reflections ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, trade_id INTEGER, time INTEGER, '
'outcome TEXT, mistake_type TEXT, lesson TEXT, tags TEXT, raw_json TEXT);')
db_exec('CREATE TABLE IF NOT EXISTS playbook ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, version INTEGER, created_at INTEGER, '
'summary TEXT, do_rules TEXT, avoid_rules TEXT, params_json TEXT, based_on_trades INTEGER);')
db_exec('CREATE TABLE IF NOT EXISTS news_cache ('
'id INTEGER PRIMARY KEY AUTOINCREMENT, time INTEGER, url TEXT UNIQUE, '
'title TEXT, desc TEXT, source TEXT, query TEXT);')
def db_exec(sql, *params):
try:
return DBExec(sql, *params) if params else DBExec(sql)
except Exception as e:
Log("DB执行失败 | {} | {}".format(sql[:100], str(e)), "#FF0000")
return None
def db_query(sql, *params):
res = db_exec(sql, *params)
cols, vals = _extract_cols_values(res)
out = []
for row in vals:
out.append({cols[i]: row[i] for i in range(min(len(cols), len(row)))})
return out
def _extract_cols_values(res):
if res is None:
return [], []
if isinstance(res, dict):
return res.get("columns", []) or [], res.get("values", []) or []
return getattr(res, "columns", []) or [], getattr(res, "values", []) or []
def db_last_id():
rows = db_query("select last_insert_rowid() as id")
return rows[0]["id"] if rows else None
# =========================
# 品种解析
# =========================
def resolve_market(store):
try:
ms = exchange.GetMarkets()
except Exception as e:
Log("GetMarkets失败: " + str(e), "#FF0000")
return
found_key, found_market = None, None
kw = INSTRUMENT_KEYWORD.upper()
# 优先级：永续 swap > 无后缀精确匹配 > 子串回退。
# 避免 "BTC_USDT" 误命中 "BTC_USDT.next_quarter" 等季度合约。
swap_hit = exact_hit = sub_hit = None
for key, market in ms.items():
ku = key.upper()
if ku == kw + ".SWAP":
swap_hit = (key, market)
break
if ku == kw:
exact_hit = exact_hit or (key, market)
elif kw in ku and sub_hit is None:
sub_hit = (key, market)
picked = swap_hit or exact_hit or sub_hit
if picked:
found_key, found_market = picked
if not found_key:
Log("未找到匹配 {} 的合约，请检查 INSTRUMENT_KEYWORD 参数".format(INSTRUMENT_KEYWORD), "#FF0000")
return
store["market"] = {
"contract": found_key,
"amountPrecision": found_market.get("AmountPrecision", 0),
"pricePrecision": found_market.get("PricePrecision", 2),
"ctVal": found_market.get("CtVal", 1) or 1,
"minQty": found_market.get("MinQty", 0) or 0,
}
save_store(store)
record_step(store, "market_resolved", {"contract": found_key}, "#00AAFF")
def resolve_period():
names = {"M15": "PERIOD_M15", "M30": "PERIOD_M30", "H1": "PERIOD_H1",
"H4": "PERIOD_H4", "D1": "PERIOD_D1"}
label = str(KLINE_PERIOD_LABEL).upper() if "KLINE_PERIOD_LABEL" in globals() else "H1"
name = names.get(label, "PERIOD_H1")
return globals().get(name, globals().get("PERIOD_H1"))
KLINE_PERIOD = None # 在 main 首次使用前惰性解析，避免参数注入时序问题
def get_kline_period():
global KLINE_PERIOD
if KLINE_PERIOD is None:
KLINE_PERIOD = resolve_period()
return KLINE_PERIOD
def get_records(contract, limit):
try:
bars = exchange.GetRecords(contract, get_kline_period()) or []
return bars[-limit:]
except Exception as e:
Log("GetRecords失败 | {} | {}".format(contract, str(e)), "#FFAA00")
return []
# =========================
# 指标层 (纯 Python 手写，不依赖平台TA库签名)
# =========================
def ema_full_series(values, period):
if not values:
return []
k = 2.0 / (period + 1)
out = [values[0]]
for v in values[1:]:
out.append(out[-1] * (1 - k) + v * k)
return out
def latest_ema(values, period):
series = ema_full_series(values, period)
return series[-1] if series else 0
def latest_rsi(values, period=14):
if len(values) < period + 1:
return 50.0
deltas = [values[i] - values[i - 1] for i in range(1, len(values))]
gains = [max(d, 0) for d in deltas]
losses = [max(-d, 0) for d in deltas]
avg_gain = mean(gains[:period])
avg_loss = mean(losses[:period])
for i in range(period, len(deltas)):
avg_gain = (avg_gain * (period - 1) + gains[i]) / period
avg_loss = (avg_loss * (period - 1) + losses[i]) / period
if avg_loss == 0:
return 100.0
rs = avg_gain / avg_loss
return 100 - 100 / (1 + rs)
def latest_macd(values, fast=12, slow=26, signal=9):
if len(values) < slow + signal:
return 0.0, 0.0, 0.0, 0.0
fast_series = ema_full_series(values, fast)
slow_series = ema_full_series(values, slow)
dif_series = [f - s for f, s in zip(fast_series, slow_series)]
dea_series = ema_full_series(dif_series, signal)
dif = dif_series[-1]
dea = dea_series[-1]
hist = dif - dea
hist_prev = (dif_series[-2] - dea_series[-2]) if len(dif_series) > 1 else hist
return dif, dea, hist, hist_prev
def latest_boll(values, period=20, mult=2):
window = values[-period:] if len(values) >= period else values
mid = mean(window)
sd = std(window)
return mid + mult * sd, mid, mid - mult * sd
def latest_atr(records, period=14):
if len(records) < 2:
return 0.0
trs = []
for i in range(1, len(records)):
h, l, pc = records[i]["High"], records[i]["Low"], records[i - 1]["Close"]
trs.append(max(h - l, abs(h - pc), abs(l - pc)))
window = trs[-period:] if len(trs) >= period else trs
return mean(window) if window else 0.0
def latest_kdj(records, n=9, k_period=3, d_period=3):
if len(records) < n:
return 50.0, 50.0, 50.0
closes = [r["Close"] for r in records]
highs = [r["High"] for r in records]
lows = [r["Low"] for r in records]
k, d = 50.0, 50.0
for i in range(n - 1, len(records)):
hh = max(highs[i - n + 1:i + 1])
ll = min(lows[i - n + 1:i + 1])
rsv = (closes[i] - ll) / (hh - ll) * 100 if hh > ll else 50.0
k = (k_period - 1) / k_period * k + 1 / k_period * rsv
d = (d_period - 1) / d_period * d + 1 / d_period * k
j = 3 * k - 2 * d
return k, d, j
def pct_change(closes, n):
if len(closes) <= n:
return 0.0
a, b = closes[-1 - n], closes[-1]
return (b - a) / a * 100 if a else 0.0
def build_indicator_snapshot(records):
if not records or len(records) < 60:
return None
closes = [r["Close"] for r in records]
price = closes[-1]
ema_fast = latest_ema(closes, 20)
ema_slow = latest_ema(closes, 60)
dif, dea, hist, hist_prev = latest_macd(closes, 12, 26, 9)
rsi_v = latest_rsi(closes, 14)
boll_upper, boll_mid, boll_lower = latest_boll(closes, 20, 2)
atr_v = latest_atr(records, 14)
k_v, d_v, j_v = latest_kdj(records, 9, 3, 3)
vols = [r.get("Volume", 0) or 0 for r in records[-20:]]
vol_ma20 = mean(vols) if vols else 0
vol_now = records[-1].get("Volume", 0) or 0
vol_ratio = vol_now / vol_ma20 if vol_ma20 > 0 else 1.0
band_width = boll_upper - boll_lower
band_pos = (price - boll_lower) / band_width if band_width > 0 else 0.5
return {
"price": round(price, 6),
"trend": {
"ema_fast": round(ema_fast, 6),
"ema_slow": round(ema_slow, 6),
"ema_bias_pct": round((ema_fast - ema_slow) / ema_slow * 100, 3) if ema_slow else 0,
"macd_dif": round(dif, 6),
"macd_dea": round(dea, 6),
"macd_hist": round(hist, 6),
"macd_hist_rising": bool(hist > hist_prev),
},
"momentum": {
"rsi14": round(rsi_v, 2),
"kdj_k": round(k_v, 2),
"kdj_d": round(d_v, 2),
"kdj_j": round(j_v, 2),
},
"volatility": {
"boll_upper": round(boll_upper, 6),
"boll_mid": round(boll_mid, 6),
"boll_lower": round(boll_lower, 6),
"band_position": round(clamp(band_pos, 0, 1), 3),
"atr": round(atr_v, 6),
"atr_pct": round(atr_v / price * 100, 3) if price else 0,
},
"volume": {"vol_ratio_to_ma20": round(vol_ratio, 3)},
"price_action": {
"pct_change_1bar": round(pct_change(closes, 1), 3),
"pct_change_6bar": round(pct_change(closes, 6), 3),
"pct_change_24bar": round(pct_change(closes, 24), 3),
},
"time": int(time.time()),
}
# =========================
# 新闻层 (BraveSearch)
# =========================
def fetch_instrument_news(store):
if not BRAVE_KEY:
return []
terms = [t.strip() for t in str(NEWS_QUERY_TERMS).split(",") if t.strip()]
queries = [
'"{}" 最新消息 价格走势'.format(INSTRUMENT_DISPLAY_NAME),
'"{}" 行情分析'.format(INSTRUMENT_DISPLAY_NAME),
]
for t in terms[:4]:
queries.append('"{}" {}'.format(INSTRUMENT_DISPLAY_NAME, t))
news = brave_news_search(queries[:6], 4)
for item in news:
db_exec('INSERT OR IGNORE INTO news_cache(time, url, title, desc, source, query) values(?,?,?,?,?,?)',
int(time.time()), item.get("url", ""), item.get("title", ""),
item.get("desc", ""), item.get("source", ""), item.get("query", ""))
record_step(store, "news_refreshed", {"count": len(news)}, "#00AAFF")
return news
def brave_news_search(queries, count):
out, seen = [], {}
for q in queries:
try:
url = ("https://api.search.brave.com/res/v1/news/search?q=" +
urllib.parse.quote(q) + "&count=" + str(count) + "&freshness=pd")
raw = http_request(url, "GET", {"X-Subscription-Token": BRAVE_KEY, "Accept": "application/json"}, None, 20)
data = json.loads(raw)
for item in data.get("results", []):
link = item.get("url", "") or item.get("title", "")
if not link or link in seen:
continue
seen[link] = True
source = item.get("profile", {}) or {}
out.append({
"title": item.get("title", "") or "",
"desc": item.get("description", "") or "",
"url": item.get("url", "") or "",
"age": item.get("age", "") or "",
"source": source.get("name", "") or "",
"query": q,
})
except Exception as e:
Log("BraveSearch失败 | {} | {}".format(q, str(e)), "#FFAA00")
Sleep(800)
return out
# =========================
# Playbook 经验库(进化载体)
# =========================
def default_playbook_params():
return {
"min_confidence": MIN_CONFIDENCE,
"base_pos_pct": BASE_POS_PCT,
"hard_stop_pct": HARD_STOP_PCT,
"trail_activate_pct": TRAIL_ACTIVATE_PCT,
"trail_giveback_pct": TRAIL_GIVEBACK_PCT,
}
def load_or_init_playbook(store):
rows = db_query("select * from playbook order by version desc limit 1")
if rows:
row = rows[0]
store["current_playbook"] = {
"version": row.get("version", 1),
"summary": row.get("summary", ""),
"do_rules": safe_json_loads(row.get("do_rules"), []),
"avoid_rules": safe_json_loads(row.get("avoid_rules"), []),
"params": safe_json_loads(row.get("params_json"), default_playbook_params()),
"based_on_trades": row.get("based_on_trades", 0),
}
else:
params = default_playbook_params()
now = int(time.time())
db_exec('INSERT INTO playbook(version, created_at, summary, do_rules, avoid_rules, params_json, based_on_trades) '
'values(?,?,?,?,?,?,?)',
1, now, "初始版本，尚无交易经验，按基础参数运行。",
json.dumps([], ensure_ascii=False), json.dumps([], ensure_ascii=False),
json.dumps(params, ensure_ascii=False), 0)
store["current_playbook"] = {
"version": 1, "summary": "初始版本，尚无交易经验，按基础参数运行。",
"do_rules": [], "avoid_rules": [], "params": params, "based_on_trades": 0,
}
save_store(store)
def playbook_param(store, key, config_default, lo, hi):
v = store.get("current_playbook", {}).get("params", {}).get(key, config_default)
try:
v = float(v)
except Exception:
v = config_default
return clamp(v, lo, hi)
# =========================
# 决策层
# =========================
def run_decision_cycle(store):
market = store.get("market")
if not market:
return
records = get_records(market["contract"], KLINE_LIMIT)
if not records or len(records) < 60:
return
decision = make_decision(store, records)
if not decision:
return
ticker = None
try:
ticker = exchange.GetTicker(market["contract"])
except Exception as e:
Log("GetTicker失败: " + str(e), "#FFAA00")
return
apply_decision(store, decision, ticker)
def make_decision(store, records):
snapshot = build_indicator_snapshot(records)
if not snapshot:
return None
store["indicator_snapshot"] = snapshot
db_exec('INSERT INTO indicator_snapshot(time, price, snapshot_json) values(?,?,?)',
snapshot["time"], snapshot["price"], json.dumps(snapshot, ensure_ascii=False))
news = store.get("news_cache_recent", [])
playbook = store.get("current_playbook", {})
pos = _G(POSITION_STATE_KEY) or {}
if not LLM_API_KEY:
record_step(store, "decision_skipped_no_llm", {}, "#FFAA00")
return None
schema = {
"action": "open_long|open_short|add_long|add_short|reduce|close|hold",
"confidence": 0,
"technical_view": "bullish|bearish|neutral|conflicted",
"news_view": "bullish|bearish|neutral|no_data",
"reason": "",
"key_indicators_cited": [""],
"suggested_stop_pct": 0.0,
"risk_notes": "",
}
prompt = "\n".join([
"你是{}单品种AI交易决策引擎。".format(INSTRUMENT_DISPLAY_NAME),
"当前playbook经验(do_rules必须遵守，avoid_rules必须避免):",
json.dumps(playbook, ensure_ascii=False),
"当前持仓状态(空字典表示无持仓):",
json.dumps(pos, ensure_ascii=False),
"多指标快照:",
json.dumps(snapshot, ensure_ascii=False),
"近期相关新闻(可能为空):",
json.dumps(news[:10], ensure_ascii=False),
"输出schema:",
json.dumps(schema, ensure_ascii=False),
"要求：",
"0. 综合趋势/动量/波动率/量能多个维度，不要只依赖单一指标下结论。",
"1. 只有technical_view与news_view不冲突(news_view为neutral/no_data，或方向与technical_view一致)才能给出open_long/open_short/add_long/add_short。",
"2. 如果news_view方向与technical_view相反，action必须是hold或close。",
"3. 已有持仓且趋势明显反转时，action应为close，不要硬扛。",
"4. confidence必须真实反映把握程度(0-100)，不要习惯性给高分，把握不足给低分。",
"5. reason和risk_notes必须使用中文，并结合playbook中的do_rules/avoid_rules说明理由。",
"6. 字段名必须和schema完全一致，不要新增或改名字段。",
])
data = call_llm_json("你是严谨的量化交易决策引擎，只输出JSON，所有中文字段必须使用中文，不要输出markdown。", prompt, 1200)
if not data:
return None
decision = sanitize_decision(data)
db_exec('INSERT INTO decisions(time, playbook_version, action, direction, confidence, technical_view, news_view, '
'reason, indicators_json, news_json, executed) values(?,?,?,?,?,?,?,?,?,?,0)',
int(time.time()), playbook.get("version", 0), decision["action"], decision["direction"],
decision["confidence"], decision["technical_view"], decision["news_view"], decision["reason"],
json.dumps(snapshot, ensure_ascii=False), json.dumps(news[:10], ensure_ascii=False))
decision["id"] = db_last_id()
store.setdefault("signals", []).insert(0, {
"time": int(time.time()), "action": decision["action"], "confidence": decision["confidence"],
"technical_view": decision["technical_view"], "news_view": decision["news_view"],
"reason": decision["reason"],
})
store["signals"] = store["signals"][:50]
record_step(store, "decision_made", {
"action": decision["action"], "confidence": decision["confidence"],
"technical_view": decision["technical_view"], "news_view": decision["news_view"],
}, "#00AAFF")
return decision
def sanitize_decision(data):
action = str(data.get("action", "hold") or "hold").lower()
valid_actions = ["open_long", "open_short", "add_long", "add_short", "reduce", "close", "hold"]
if action not in valid_actions:
action = "hold"
direction = "long" if "long" in action else ("short" if "short" in action else "")
return {
"action": action,
"direction": direction,
"confidence": normalize_confidence(data.get("confidence", 0)),
"technical_view": data.get("technical_view", "neutral") or "neutral",
"news_view": data.get("news_view", "no_data") or "no_data",
"reason": data.get("reason", "") or "",
"key_indicators_cited": as_list(data.get("key_indicators_cited")),
"suggested_stop_pct": safe_float(data.get("suggested_stop_pct", 0)),
"risk_notes": data.get("risk_notes", "") or "",
}
# =========================
# 执行层
# =========================
def apply_decision(store, decision, ticker):
if not ticker:
return
action = decision["action"]
pos = _G(POSITION_STATE_KEY) or {}
has_pos = bool(pos and pos.get("qty", 0) > 0)
min_conf = playbook_param(store, "min_confidence", MIN_CONFIDENCE, MIN_CONFIDENCE_FLOOR, MIN_CONFIDENCE_CEIL)
if action in ("open_long", "open_short") and not has_pos:
if decision["confidence"] < min_conf:
record_step(store, "decision_skip_low_confidence",
{"action": action, "confidence": decision["confidence"], "min": min_conf}, "#999999")
return
open_position(store, decision, ticker, add_mode=False)
elif action in ("open_long", "open_short") and has_pos and pos.get("direction") != decision["direction"]:
close_position(store, ticker, "reverse_signal")
Sleep(1000)
if decision["confidence"] >= min_conf:
open_position(store, decision, ticker, add_mode=False)
elif action in ("add_long", "add_short") and has_pos and pos.get("direction") == decision["direction"]:
if decision["confidence"] < min_conf:
return
if pos.get("add_count", 0) >= MAX_TOTAL_ADD_COUNT:
record_step(store, "add_blocked_max_count", {"add_count": pos.get("add_count", 0)}, "#FFAA00")
return
open_position(store, decision, ticker, add_mode=True)
elif action in ("close", "reduce") and has_pos:
if action == "reduce":
reduce_position(store, ticker, 0.5, "ai_reduce")
else:
close_position(store, ticker, "ai_decision")
else:
record_step(store, "decision_hold", {"action": action, "confidence": decision["confidence"]}, "#999999")
def open_position(store, decision, ticker, add_mode):
market = store.get("market")
if not market or not ticker:
return False
if TRADE_MODE != "trade":
record_step(store, "trade_notify", {
"action": decision["action"], "direction": decision["direction"],
"confidence": decision["confidence"], "reason": decision["reason"],
}, "#00CC66")
return True
acc = None
try:
acc = exchange.GetAccount()
except Exception as e:
Log("GetAccount失败: " + str(e), "#FF0000")
return False
if not acc or not ticker or ticker.get("Last", 0) <= 0:
return False
equity = get_equity(acc)
base_pos_pct = playbook_param(store, "base_pos_pct", BASE_POS_PCT, BASE_POS_PCT_FLOOR, BASE_POS_PCT_CEIL)
size_pct = base_pos_pct * clamp(decision["confidence"] / 100.0, 0.4, 1.0)
qty = _N(equity * size_pct * LEVERAGE / ticker["Last"] / market["ctVal"], market["amountPrecision"])
if qty <= 0 or qty < market["minQty"]:
record_step(store, "trade_blocked", {"reason": "数量过小", "qty": qty}, "#FFAA00")
return False
pos = _G(POSITION_STATE_KEY) or {}
cur_qty = pos.get("qty", 0) if add_mode else 0
max_qty = _N(equity * MAX_SINGLE_POS_PCT * LEVERAGE / ticker["Last"] / market["ctVal"], market["amountPrecision"])
if cur_qty + qty > max_qty:
qty = _N(max(0, max_qty - cur_qty), market["amountPrecision"])
if qty <= 0 or qty < market["minQty"]:
record_step(store, "trade_blocked", {"reason": "已达单标的仓位上限"}, "#FFAA00")
return False
side = "buy" if decision["direction"] == "long" else "sell"
oid = None
try:
oid = exchange.CreateOrder(market["contract"], side, -1, qty)
except Exception as e:
Log("下单异常: " + str(e), "#FF0000")
if not oid:
record_step(store, "trade_failed", {"side": side, "qty": qty}, "#FF0000")
return False
hard_stop_pct = playbook_param(store, "hard_stop_pct", HARD_STOP_PCT, HARD_STOP_PCT_MIN, HARD_STOP_PCT_MAX)
now = int(time.time())
if add_mode and pos:
new_qty = pos.get("qty", 0) + qty
avg_entry = ((pos.get("entry", ticker["Last"]) * pos.get("qty", 0) + ticker["Last"] * qty) / new_qty
if new_qty > 0 else ticker["Last"])
pos["qty"] = new_qty
pos["entry"] = avg_entry
pos["add_count"] = pos.get("add_count", 0) + 1
else:
pos = {
"direction": decision["direction"], "qty": qty, "entry": ticker["Last"],
"opened_at": now, "add_count": 0, "peakPnlPct": 0, "trailActive": False,
"decision_id": decision.get("id"),
}
pos["stop"] = (pos["entry"] * (1 - hard_stop_pct / 100.0) if pos["direction"] == "long"
else pos["entry"] * (1 + hard_stop_pct / 100.0))
_G(POSITION_STATE_KEY, pos)
if decision.get("id"):
db_exec("UPDATE decisions SET executed=1 WHERE id=?", decision["id"])
record_step(store, "trade_opened", {
"direction": pos["direction"], "qty": qty, "price": ticker["Last"],
"add_mode": add_mode, "hard_stop_pct": hard_stop_pct,
}, "#00CC66")
return True
def close_position(store, ticker, reason):
pos = _G(POSITION_STATE_KEY)
if not pos or not pos.get("qty"):
return False
market = store.get("market")
if not market:
return False
if TRADE_MODE == "trade":
side = "closebuy" if pos["direction"] == "long" else "closesell"
oid = None
try:
oid = exchange.CreateOrder(market["contract"], side, -1, pos["qty"])
except Exception as e:
Log("平仓异常: " + str(e), "#FF0000")
if not oid:
record_step(store, "close_failed", {"reason": reason}, "#FF0000")
return False
exit_price = ticker["Last"] if ticker else pos.get("entry", 0)
pnl_pct = (exit_price - pos["entry"]) / pos["entry"] * 100 * (1 if pos["direction"] == "long" else -1)
now = int(time.time())
db_exec('INSERT INTO trades(open_time, close_time, direction, entry_price, exit_price, qty, pnl_pct, '
'close_reason, decision_id) values(?,?,?,?,?,?,?,?,?)',
pos.get("opened_at", now), now, pos["direction"], pos["entry"], exit_price,
pos["qty"], round(pnl_pct, 3), reason, pos.get("decision_id"))
trade_id = db_last_id()
record_step(store, "trade_closed", {
"direction": pos["direction"], "pnl_pct": round(pnl_pct, 3), "reason": reason,
}, "#00CC66" if pnl_pct > 0 else "#FF6666")
_G(POSITION_STATE_KEY, None)
if trade_id:
reflect_on_closed_trade(store, trade_id)
store["closed_trade_count_since_evolution"] = store.get("closed_trade_count_since_evolution", 0) + 1
save_store(store)
maybe_run_evolution(store)
return True
def reduce_position(store, ticker, ratio, reason):
pos = _G(POSITION_STATE_KEY)
if not pos or not pos.get("qty"):
return False
market = store.get("market")
if not market:
return False
cut = _N(max(0, pos["qty"] * clamp(ratio, 0.1, 1.0)), market["amountPrecision"])
if cut <= 0 or cut < market.get("minQty", 0):
return False
if cut >= pos["qty"]:
return close_position(store, ticker, reason)
if TRADE_MODE == "trade":
side = "closebuy" if pos["direction"] == "long" else "closesell"
oid = None
try:
oid = exchange.CreateOrder(market["contract"], side, -1, cut)
except Exception as e:
Log("减仓异常: " + str(e), "#FF0000")
if not oid:
record_step(store, "reduce_failed", {"reason": reason}, "#FF0000")
return False
pos["qty"] = _N(pos["qty"] - cut, market["amountPrecision"])
_G(POSITION_STATE_KEY, pos)
record_step(store, "position_reduced", {"cut": cut, "left": pos["qty"], "reason": reason}, "#FFAA00")
return True
def monitor_position(store):
pos = _G(POSITION_STATE_KEY)
if not pos or not pos.get("qty"):
return
market = store.get("market")
if not market:
return
ticker = None
try:
ticker = exchange.GetTicker(market["contract"])
except Exception as e:
Log("GetTicker失败: " + str(e), "#FFAA00")
return
if not ticker or ticker.get("Last", 0) <= 0:
return
direction = pos["direction"]
pnl_pct = (ticker["Last"] - pos["entry"]) / pos["entry"] * 100 * (1 if direction == "long" else -1)
if pnl_pct > pos.get("peakPnlPct", 0):
pos["peakPnlPct"] = pnl_pct
trail_activate = playbook_param(store, "trail_activate_pct", TRAIL_ACTIVATE_PCT,
TRAIL_ACTIVATE_PCT_MIN, TRAIL_ACTIVATE_PCT_MAX)
trail_giveback = playbook_param(store, "trail_giveback_pct", TRAIL_GIVEBACK_PCT,
TRAIL_GIVEBACK_PCT_MIN, TRAIL_GIVEBACK_PCT_MAX)
if not pos.get("trailActive") and pos.get("peakPnlPct", 0) >= trail_activate:
pos["trailActive"] = True
hard_stop = ticker["Last"] <= pos["stop"] if direction == "long" else ticker["Last"] >= pos["stop"]
trail_hit = False
if pos.get("trailActive"):
giveback = pos.get("peakPnlPct", 0) - pnl_pct
allowed = max(1.0, pos.get("peakPnlPct", 0) * trail_giveback / 100.0)
trail_hit = giveback >= allowed
if hard_stop or trail_hit:
close_position(store, ticker, "hard_stop" if hard_stop else "trailing_stop")
else:
_G(POSITION_STATE_KEY, pos)
def get_equity(acc):
return acc.get("Equity") or acc.get("Balance") or acc.get("Stocks") or 0
# =========================
# 复盘层
# =========================
def reflect_on_closed_trade(store, trade_id):
if not LLM_API_KEY:
return
rows = db_query("select * from trades where id=?", trade_id)
if not rows:
return
trade = rows[0]
entry_decision = None
if trade.get("decision_id"):
drows = db_query("select * from decisions where id=?", trade["decision_id"])
entry_decision = drows[0] if drows else None
schema = {
"outcome": "win|loss|breakeven",
"mistake_type": "none|chased_move|ignored_news_conflict|stop_too_tight|stop_too_wide|"
"overleveraged|held_too_long|exited_too_early|other",
"lesson": "",
"tags": [""],
}
prompt = "\n".join([
"对以下已平仓交易进行复盘：",
"交易记录:", json.dumps(trade, ensure_ascii=False),
"入场时的决策依据:", json.dumps(entry_decision, ensure_ascii=False) if entry_decision else "无记录",
"输出schema:", json.dumps(schema, ensure_ascii=False),
"要求：",
"1. outcome根据pnl_pct判断，明显为正=win，明显为负=loss，接近0=breakeven。",
"2. lesson必须是具体、可执行的中文经验总结，不要空泛(例如'RSI超过75追多容易被套，应等回调')。",
"3. mistake_type只在确实存在对应问题时才选择非none选项。",
"4. 字段名必须和schema完全一致。",
])
data = call_llm_json("你是交易复盘教练，只输出JSON，所有文本字段必须使用中文。", prompt, 900)
if not data:
return
now = int(time.time())
db_exec('INSERT INTO reflections(trade_id, time, outcome, mistake_type, lesson, tags, raw_json) '
'values(?,?,?,?,?,?,?)',
trade_id, now, data.get("outcome", ""), data.get("mistake_type", "none"),
data.get("lesson", ""), json.dumps(as_list(data.get("tags")), ensure_ascii=False),
json.dumps(data, ensure_ascii=False))
db_exec('UPDATE trades SET reflected=1 WHERE id=?', trade_id)
record_step(store, "reflection_saved", {
"trade_id": trade_id, "outcome": data.get("outcome"),
"lesson": short_text(data.get("lesson", ""), 80),
}, "#00AAFF")
# =========================
# 进化层 (核心：自进化闭环)
# =========================
def maybe_run_evolution(store):
if store.get("evolution_paused"):
return
total_reflected = db_query("select count(*) as c from reflections")
total_reflected = total_reflected[0]["c"] if total_reflected else 0
if total_reflected < EVOLUTION_MIN_TRADES:
return
now = int(time.time())
closed_since = store.get("closed_trade_count_since_evolution", 0)
trade_due = closed_since >= EVOLUTION_TRADE_INTERVAL
time_due = now - store.get("last_evolution_check", 0) > EVOLUTION_TIME_INTERVAL_S
if trade_due or time_due:
run_evolution(store)
def run_evolution(store, forced=False):
rows = db_query(
"select r.*, t.direction as t_direction, t.pnl_pct as t_pnl_pct, t.close_reason as t_close_reason "
"from reflections r left join trades t on r.trade_id = t.id order by r.time desc limit ?",
REFLECTION_LOOKBACK)
if not rows:
if not forced:
return
record_step(store, "evolution_skipped", {"reason": "暂无复盘记录"}, "#FFAA00")
return
if not LLM_API_KEY:
record_step(store, "evolution_skipped", {"reason": "LLM_API_KEY未配置"}, "#FFAA00")
return
stats = compute_trade_stats()
playbook = store.get("current_playbook", {})
schema = {
"summary": "",
"do_rules": [""],
"avoid_rules": [""],
"param_suggestions": {
"min_confidence": 0,
"base_pos_pct": 0.0,
"hard_stop_pct": 0.0,
"trail_activate_pct": 0.0,
"trail_giveback_pct": 0.0,
},
"confidence_in_update": 0,
"reasoning": "",
}
prompt = "\n".join([
"你是交易系统的自我进化复盘员，任务是基于历史交易复盘更新交易playbook(经验库)。",
"当前playbook:", json.dumps(playbook, ensure_ascii=False),
"整体统计:", json.dumps(stats, ensure_ascii=False),
"最近复盘记录:", json.dumps(rows[:REFLECTION_LOOKBACK], ensure_ascii=False),
"输出schema:", json.dumps(schema, ensure_ascii=False),
"要求：",
"1. do_rules/avoid_rules必须是具体、可执行的中文规则，例如'RSI>70且新闻利空时不要开多'。",
"2. param_suggestions只在有充分证据支持时才调整，避免频繁大幅震荡；证据不足时原样返回当前值。",
"3. summary是对当前策略状态的整体中文总结，控制在150字以内。",
"4. 字段名必须和schema完全一致，不要新增或改名。",
])
data = call_llm_json("你是量化交易复盘与策略进化专家，只输出JSON，所有文本字段必须使用中文。", prompt, 1600)
if not data:
record_step(store, "evolution_failed", {"reason": "LLM返回空"}, "#FFAA00")
return
old_params = playbook.get("params", default_playbook_params())
suggested = data.get("param_suggestions", {}) or {}
new_params = dict(old_params)
bounds = {
"min_confidence": (MIN_CONFIDENCE_FLOOR, MIN_CONFIDENCE_CEIL),
"base_pos_pct": (BASE_POS_PCT_FLOOR, BASE_POS_PCT_CEIL),
"hard_stop_pct": (HARD_STOP_PCT_MIN, HARD_STOP_PCT_MAX),
"trail_activate_pct": (TRAIL_ACTIVATE_PCT_MIN, TRAIL_ACTIVATE_PCT_MAX),
"trail_giveback_pct": (TRAIL_GIVEBACK_PCT_MIN, TRAIL_GIVEBACK_PCT_MAX),
}
for k, (lo, hi) in bounds.items():
if k in suggested:
try:
new_params[k] = round(clamp(float(suggested[k]), lo, hi), 4)
except Exception:
pass
new_version = playbook.get("version", 0) + 1
now = int(time.time())
do_rules = as_list(data.get("do_rules"))[:12]
avoid_rules = as_list(data.get("avoid_rules"))[:12]
summary = data.get("summary", "") or ""
db_exec('INSERT INTO playbook(version, created_at, summary, do_rules, avoid_rules, params_json, based_on_trades) '
'values(?,?,?,?,?,?,?)',
new_version, now, summary, json.dumps(do_rules, ensure_ascii=False),
json.dumps(avoid_rules, ensure_ascii=False), json.dumps(new_params, ensure_ascii=False), len(rows))
store["current_playbook"] = {
"version": new_version, "summary": summary, "do_rules": do_rules,
"avoid_rules": avoid_rules, "params": new_params, "based_on_trades": len(rows),
}
store["closed_trade_count_since_evolution"] = 0
store["last_evolution_check"] = now
save_store(store)
record_step(store, "evolution_done", {
"version": new_version, "summary": summary,
"old_params": old_params, "new_params": new_params, "based_on_trades": len(rows),
}, "#00CC66")
def compute_trade_stats():
rows = db_query("select pnl_pct from trades where close_time is not null order by close_time desc limit 100")
pnls = [r["pnl_pct"] for r in rows if r.get("pnl_pct") is not None]
if not pnls:
return {"total": 0, "win_rate": 0, "avg_pnl_pct": 0}
wins = [p for p in pnls if p > 0]
losses = [p for p in pnls if p <= 0]
return {
"total": len(pnls),
"win_rate": round(len(wins) / len(pnls), 3),
"avg_pnl_pct": round(mean(pnls), 3),
"avg_win_pct": round(mean(wins), 3) if wins else 0,
"avg_loss_pct": round(mean(losses), 3) if losses else 0,
}
# =========================
# Dashboard
# =========================
def show_dashboard(store):
market = store.get("market") or {}
playbook = store.get("current_playbook", {})
pos = _G(POSITION_STATE_KEY) or {}
stats = compute_trade_stats()
header = "{} 单品种AI自进化交易系统 | {} | mode:{} | playbook v{}\n".format(
INSTRUMENT_DISPLAY_NAME, time.strftime("%Y-%m-%d %H:%M:%S"), TRADE_MODE, playbook.get("version", 0))
overview_rows = [
["合约", market.get("contract", "-")],
["总交易数", stats.get("total", 0)],
["胜率", "{}%".format(round(stats.get("win_rate", 0) * 100, 1))],
["平均盈亏%", stats.get("avg_pnl_pct", 0)],
["当前持仓", "{}/{}".format(pos.get("direction", "-"), fmt(pos.get("qty", 0))) if pos else "无"],
["累计加仓次数", pos.get("add_count", 0) if pos else 0],
["距下次进化", "{}/{}笔".format(store.get("closed_trade_count_since_evolution", 0), EVOLUTION_TRADE_INTERVAL)],
["进化状态", "已暂停" if store.get("evolution_paused") else "运行中"],
]
overview = make_table("系统概览", ["项目", "数值"], overview_rows)
snap = store.get("indicator_snapshot") or {}
ind_rows = []
if snap:
t, m = snap.get("trend", {}), snap.get("momentum", {})
v, vo, pa = snap.get("volatility", {}), snap.get("volume", {}), snap.get("price_action", {})
ind_rows = build_indicator_rows(snap, t, m, v, vo, pa)
if not ind_rows:
ind_rows = [["-", "-", "-", "-", "-", "-", "-"]]
indicator_table = make_table(
"📊 多指标快照",
["维度", "指标①", "数值①", "指标②", "数值②", "指标③", "数值③"],
ind_rows)
signal_rows = []
for s in store.get("signals", [])[:8]:
signal_rows.append([
time.strftime("%H:%M:%S", time.localtime(s.get("time", 0))), s.get("action"), s.get("confidence"),
s.get("technical_view"), s.get("news_view"), short_text(s.get("reason", ""), 40),
])
if not signal_rows:
signal_rows = [["-", "-", "-", "-", "-", "暂无"]]
signals_table = make_table("最近决策", ["时间", "动作", "置信度", "技术面", "消息面", "理由"], signal_rows)
trade_rows = []
for t in db_query("select * from trades order by close_time desc limit 8"):
trade_rows.append([
time.strftime("%H:%M:%S", time.localtime(t.get("close_time") or 0)), t.get("direction"),
fmt(t.get("entry_price")), fmt(t.get("exit_price")), fmt(t.get("pnl_pct")), t.get("close_reason"),
])
if not trade_rows:
trade_rows = [["-", "-", "-", "-", "-", "暂无"]]
trades_table = make_table("最近交易", ["平仓时间", "方向", "入场", "出场", "盈亏%", "原因"], trade_rows)
ref_rows = []
for r in db_query("select * from reflections order by time desc limit 6"):
ref_rows.append([
time.strftime("%H:%M:%S", time.localtime(r.get("time") or 0)), r.get("outcome"),
r.get("mistake_type"), short_text(r.get("lesson", ""), 60),
])
if not ref_rows:
ref_rows = [["-", "-", "-", "暂无"]]
reflections_table = make_table("最近复盘", ["时间", "结果", "问题类型", "经验"], ref_rows)
playbook_rows = [
["版本", playbook.get("version", 0)],
["摘要", short_text(playbook.get("summary", ""), 80)],
["Do规则", " | ".join([short_text(x, 30) for x in playbook.get("do_rules", [])[:5]]) or "-"],
["Avoid规则", " | ".join([short_text(x, 30) for x in playbook.get("avoid_rules", [])[:5]]) or "-"],
["当前参数", json.dumps(playbook.get("params", {}), ensure_ascii=False)],
]
playbook_table = make_table("当前Playbook(经验库)", ["项目", "内容"], playbook_rows)
step_rows = []
for row in store.get("step_logs", [])[:8]:
step_rows.append([
time.strftime("%H:%M:%S", time.localtime(row.get("time", 0))), row.get("step", ""),
short_text(json.dumps(row.get("data", {}), ensure_ascii=False), 90),
])
if not step_rows:
step_rows = [["-", "-", "暂无"]]
steps_table = make_table("最近步骤", ["时间", "步骤", "内容"], step_rows)
LogStatus(
header +
status_table(overview) +
status_table(indicator_table) +
status_table(signals_table) +
status_table(trades_table) +
status_table(reflections_table) +
status_table(playbook_table) +
status_table(steps_table)
)
def build_indicator_rows(snap, t, m, v, vo, pa):
price = snap.get("price")
ema_f, ema_s = t.get("ema_fast"), t.get("ema_slow")
bias = t.get("ema_bias_pct", 0)
# 趋势方向标记：快线在慢线上方且价格在快线上方 -> 多头排列
trend_tag = "多头↑" if (num(ema_f) > num(ema_s) and num(price) > num(ema_f)) else \
("空头↓" if (num(ema_f) < num(ema_s) and num(price) < num(ema_f)) else "震荡→")
hist = t.get("macd_hist")
macd_tag = "红↑" if num(hist) > 0 else "绿↓"
rsi = m.get("rsi14")
rsi_tag = rsi_state(rsi)
k, d = m.get("kdj_k"), m.get("kdj_d")
kdj_tag = "金叉↑" if num(k) > num(d) else "死叉↓"
bp = v.get("band_position")
bp_tag = band_state(bp)
chg1 = pa.get("pct_change_1bar")
chg6 = pa.get("pct_change_6bar")
return [
["趋势",
"价格", fmt_price(price),
"EMA20", fmt_price(ema_f),
"EMA60", fmt_price(ema_s)],
["趋势",
"均线状态", trend_tag,
"快慢乖离", fmt_pct(bias),
"MACD柱", "{} {}".format(fmt_price(hist), macd_tag)],
["动量",
"RSI14", "{} {}".format(fmt_num(rsi), rsi_tag),
"KDJ_K", fmt_num(k),
"KDJ状态", "{} {}".format(fmt_num(d), kdj_tag)],
["通道",
"BOLL上", fmt_price(v.get("boll_upper")),
"BOLL下", fmt_price(v.get("boll_lower")),
"带宽位置", "{} {}".format(fmt_num(bp), bp_tag)],
["波动量能",
"ATR%", fmt_pct(v.get("atr_pct")),
"量比", fmt_num(vo.get("vol_ratio_to_ma20")),
"近1/6根%", "{} / {}".format(fmt_pct(chg1), fmt_pct(chg6))],
]
def rsi_state(rsi):
r = num(rsi)
if r >= 70:
return "超买"
if r <= 30:
return "超卖"
if r >= 55:
return "偏强"
if r <= 45:
return "偏弱"
return "中性"
def band_state(bp):
b = num(bp)
if b >= 0.8:
return "近上轨"
if b <= 0.2:
return "近下轨"
if b >= 0.55:
return "中上"
if b <= 0.45:
return "中下"
return "居中"
def num(x):
try:
return float(x)
except Exception:
return 0.0
def fmt_price(x):
v = num(x)
if abs(v) >= 1000:
return "{:,.1f}".format(v)
if abs(v) >= 1:
return "{:.3f}".format(v)
return "{:.5f}".format(v)
def fmt_num(x):
if x is None:
return "-"
return "{:.2f}".format(num(x))
def fmt_pct(x):
if x is None:
return "-"
return "{:+.2f}%".format(num(x))
def make_table(title, cols, rows):
return {"type": "table", "title": title, "cols": cols, "rows": rows}
def status_table(table):
return "\n`" + json.dumps(table, ensure_ascii=False) + "`\n"
# =========================
# 工具函数
# =========================
def call_llm_json(system_prompt, user_prompt, max_tokens):
body = json.dumps({
"model": LLM_MODEL,
"temperature": 0.1,
"max_tokens": max_tokens,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
],
})
try:
raw = http_request(LLM_BASE_URL, "POST", {
"Content-Type": "application/json",
"Authorization": "Bearer " + LLM_API_KEY,
}, body, 40)
data = json.loads(raw)
if data.get("error"):
Log("LLM错误: " + json.dumps(data["error"], ensure_ascii=False), "#FF0000")
return None
return json.loads(clean_json_text(data["choices"][0]["message"]["content"]))
except Exception as e:
Log("LLM解析失败: " + str(e), "#FFAA00")
return None
def http_request(url, method, headers, body, timeout):
data = body.encode("utf-8") if body else None
req = urllib.request.Request(url, data=data, headers=headers or {}, method=method)
resp = urllib.request.urlopen(req, timeout=timeout)
try:
return resp.read().decode("utf-8", errors="ignore")
finally:
resp.close()
def clean_json_text(text):
text = re.sub(r"<think>[\s\S]*?</think>", "", text or "").strip()
text = re.sub(r"```json\s*|```\s*", "", text).strip()
first, last = text.find("{"), text.rfind("}")
return text[first:last + 1] if first >= 0 and last > first else text
def normalize_confidence(value):
try:
v = float(value or 0)
except Exception:
return 0
if 0 < v <= 1:
v = v * 100
return int(round(max(0, min(100, v))))
def safe_float(value, default=0.0):
try:
return float(value)
except Exception:
return default
def safe_json_loads(text, default):
try:
return json.loads(text) if text else default
except Exception:
return default
def as_list(x):
if x is None:
return []
if isinstance(x, list):
return x
return [x]
def mean(arr):
return sum(arr) / len(arr) if arr else 0.0
def std(arr):
if len(arr) < 2:
return 0.0
m = mean(arr)
return math.sqrt(sum((x - m) ** 2 for x in arr) / (len(arr) - 1))
def clamp(x, lo, hi):
return max(lo, min(hi, x))
def short_text(text, n):
text = str(text or "").replace("\n", " ").replace("\r", " ")
return text[:n] + ("..." if len(text) > n else "")
def fmt(x):
try:
return round(float(x), 4)
except Exception:
return xStrategy parameters
Commands
Related strategies
Comment
All comments (0)
No data
- 1