dekun
141 lines
4.7 KiB
Python
141 lines
4.7 KiB
Python
"""斐波关键位监控：纯计算与类型判断（Gate / Binance 主站共用）。"""
|
||
|
||
from key_monitor_lib import KEY_MONITOR_AUTO_TYPES
|
||
|
||
FIB_KEY_MONITOR_TYPES = frozenset({"斐波回调0.618", "斐波回调0.786"})
|
||
KEY_MONITOR_TRADE_TYPE = "关键位监控"
|
||
|
||
FIB_RATIO_BY_TYPE = {
|
||
"斐波回调0.618": 0.618,
|
||
"斐波回调0.786": 0.786,
|
||
}
|
||
|
||
|
||
def is_fib_key_monitor_type(monitor_type):
|
||
return (monitor_type or "").strip() in FIB_KEY_MONITOR_TYPES
|
||
|
||
|
||
def fib_ratio_from_type(monitor_type):
|
||
return FIB_RATIO_BY_TYPE.get((monitor_type or "").strip())
|
||
|
||
|
||
def calc_fib_plan(direction, upper, lower, ratio):
|
||
"""
|
||
上沿 H、下沿 L（H > L）。
|
||
做多：自 H 向下回撤 ratio，E = H - ratio*(H-L)；SL=L，TP=H。
|
||
做空：自 L 向上反弹 ratio，E = L + ratio*(H-L)；SL=H，TP=L。
|
||
返回 (entry, stop_loss, take_profit) 或 None。
|
||
"""
|
||
try:
|
||
h = float(upper)
|
||
l = float(lower)
|
||
r = float(ratio)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if h <= l or r <= 0 or r >= 1:
|
||
return None
|
||
span = h - l
|
||
direction = (direction or "long").strip().lower()
|
||
if direction == "short":
|
||
entry = l + r * span
|
||
return entry, h, l
|
||
entry = h - r * span
|
||
return entry, l, h
|
||
|
||
|
||
def stored_key_signal_type(monitor_type):
|
||
"""写入 order_monitors / trade_records 的 key_signal_type（箱体/收敛/斐波/假突破/触价开仓）。"""
|
||
mt = (monitor_type or "").strip()
|
||
if mt in FIB_KEY_MONITOR_TYPES:
|
||
return mt
|
||
if mt in ("假突破", "触价开仓"):
|
||
return mt
|
||
if mt in KEY_MONITOR_AUTO_TYPES:
|
||
return mt
|
||
return None
|
||
|
||
|
||
KEY_ENTRY_REASON_BY_SIGNAL = {
|
||
"箱体突破": "关键位箱体突破",
|
||
"收敛突破": "关键位收敛突破",
|
||
"斐波回调0.618": "关键位斐波0.618",
|
||
"斐波回调0.786": "关键位斐波0.786",
|
||
"假突破": "关键位假突破",
|
||
"触价开仓": "关键位触价开仓",
|
||
"趋势回调": "趋势回调",
|
||
}
|
||
|
||
|
||
def entry_reason_from_key_signal(key_signal_type):
|
||
return KEY_ENTRY_REASON_BY_SIGNAL.get((key_signal_type or "").strip())
|
||
|
||
|
||
def key_signal_type_for_trade_record(key_signal_type, box_auto_types):
|
||
"""平仓写入 trade_records 时保留箱体/收敛/斐波/假突破来源。"""
|
||
kst = (key_signal_type or "").strip()
|
||
if kst in FIB_KEY_MONITOR_TYPES:
|
||
return kst
|
||
if kst == "假突破":
|
||
return kst
|
||
if kst == "触价开仓":
|
||
return kst
|
||
if box_auto_types and kst in box_auto_types:
|
||
return kst
|
||
return None
|
||
|
||
|
||
def backfill_missing_key_signal_types(conn, *, monitor_type: str = KEY_MONITOR_TRADE_TYPE) -> int:
|
||
"""补全历史 trade_records / order_monitors 中缺失的箱体/收敛 key_signal_type。"""
|
||
mt = (monitor_type or KEY_MONITOR_TRADE_TYPE).strip()
|
||
updated = 0
|
||
for signal in KEY_MONITOR_AUTO_TYPES:
|
||
entry_reason = KEY_ENTRY_REASON_BY_SIGNAL.get(signal)
|
||
if entry_reason:
|
||
cur = conn.execute(
|
||
"""UPDATE trade_records SET key_signal_type=?
|
||
WHERE monitor_type=? AND (key_signal_type IS NULL OR TRIM(key_signal_type)='')
|
||
AND TRIM(COALESCE(entry_reason, ''))=?""",
|
||
(signal, mt, entry_reason),
|
||
)
|
||
updated += int(cur.rowcount or 0)
|
||
rows = conn.execute(
|
||
"""SELECT id, symbol, opened_at FROM trade_records
|
||
WHERE monitor_type=? AND (key_signal_type IS NULL OR TRIM(key_signal_type)='')""",
|
||
(mt,),
|
||
).fetchall()
|
||
for row in rows:
|
||
# init_db 连接未设 row_factory，结果为 tuple
|
||
rid, sym, opened_at = row[0], row[1], row[2]
|
||
opened = (opened_at or "").strip()
|
||
for signal in KEY_MONITOR_AUTO_TYPES:
|
||
hist = conn.execute(
|
||
"""SELECT monitor_type FROM key_monitor_history
|
||
WHERE symbol=? AND monitor_type=? AND close_reason='auto_opened'
|
||
AND (?='' OR closed_at <= ?)
|
||
ORDER BY closed_at DESC LIMIT 1""",
|
||
(sym, signal, opened, opened),
|
||
).fetchone()
|
||
if not hist:
|
||
continue
|
||
conn.execute(
|
||
"UPDATE trade_records SET key_signal_type=? WHERE id=?",
|
||
(signal, rid),
|
||
)
|
||
updated += 1
|
||
break
|
||
return updated
|
||
|
||
|
||
def fib_invalidate_by_mark(direction, mark_price, upper, lower):
|
||
"""先触达止盈侧（标记价）则失效。多：mark>=H；空：mark<=L。"""
|
||
try:
|
||
m = float(mark_price)
|
||
h = float(upper)
|
||
l = float(lower)
|
||
except (TypeError, ValueError):
|
||
return False
|
||
direction = (direction or "long").strip().lower()
|
||
if direction == "short":
|
||
return m <= l
|
||
return m >= h
|