187 lines
8.0 KiB
Python
187 lines
8.0 KiB
Python
#!/usr/bin/env python3
|
|
"""GENESIS — Hephaestus Tool + Telegram Bot (Strategy F: Grid+Martingale)
|
|
CLI for Hermes. Bot for manual monitoring and override.
|
|
"""
|
|
import sys, os, json, time, logging, threading
|
|
sys.path.insert(0, str(__import__("pathlib").Path(__file__).parent))
|
|
|
|
from pathlib import Path
|
|
from datetime import datetime, timezone
|
|
import yaml, requests
|
|
|
|
CONFIG_PATH = Path(__file__).parent / "hephaestus_config.yaml"
|
|
if not CONFIG_PATH.exists():
|
|
CONFIG_PATH = Path(__file__).parents[2] / "configs" / "hephaestus_config.yaml"
|
|
with open(CONFIG_PATH, encoding="utf-8") as f:
|
|
CFG = yaml.safe_load(f)
|
|
|
|
TG_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
|
|
TG_CHAT_ID = str(CFG["telegram"]["chat_id"])
|
|
STRATEGY = CFG["strategy"]["name"]
|
|
COMMENT = CFG["strategy"]["comment"]
|
|
|
|
# Resolve safe log path (fallback to local logs/ if system dir not writable)
|
|
default_log = "/var/log/hephaestus/hephaestus_bot.log"
|
|
try:
|
|
Path(default_log).parent.mkdir(parents=True, exist_ok=True)
|
|
log_file = default_log
|
|
except Exception:
|
|
local_log_dir = Path(__file__).parents[2] / "logs" / "hephaestus"
|
|
local_log_dir.mkdir(parents=True, exist_ok=True)
|
|
log_file = str(local_log_dir / "hephaestus_bot.log")
|
|
|
|
logging.basicConfig(
|
|
filename=log_file,
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(message)s"
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
_lock = threading.Lock()
|
|
|
|
def tg_send(text):
|
|
try:
|
|
requests.post(f"https://api.telegram.org/bot{TG_TOKEN}/sendMessage",
|
|
json={"chat_id": TG_CHAT_ID, "text": text, "parse_mode": "Markdown"}, timeout=10)
|
|
except: pass
|
|
|
|
def tg_updates(offset=0):
|
|
try:
|
|
r = requests.get(f"https://api.telegram.org/bot{TG_TOKEN}/getUpdates",
|
|
params={"timeout":30,"offset":offset}, timeout=40)
|
|
return r.json().get("result",[])
|
|
except: return []
|
|
|
|
def _load():
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location("hephaestus_cycle",
|
|
Path(__file__).parent/"hephaestus_cycle.py")
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
# ── Commands ───────────────────────────────────────────────────────────────────
|
|
def do_run_tick():
|
|
try: return _load().run_cycle()
|
|
except Exception as e: return {"error":str(e)}
|
|
|
|
def do_status():
|
|
try: return _load().get_status()
|
|
except Exception as e: return {"error":str(e)}
|
|
|
|
def do_kill():
|
|
try: return _load().emergency_kill()
|
|
except Exception as e: return {"error":str(e)}
|
|
|
|
def do_enable():
|
|
try: return _load().enable_strategy()
|
|
except Exception as e: return {"error":str(e)}
|
|
|
|
def send_status_tg():
|
|
r = do_status()
|
|
if "error" in r:
|
|
tg_send(f"🔴 *{STRATEGY}*: {r['error']}"); return
|
|
enabled = r.get("enabled")
|
|
acc = r.get("account",{})
|
|
j = r.get("journal",{})
|
|
cb = r.get("circuit_breakers",{})
|
|
killed = r.get("killed_reason")
|
|
tg_send(
|
|
f"{'🔩' if enabled else '💀'} *{STRATEGY} — Grid Status*\n\n"
|
|
f"Status: {'🟢 RUNNING' if enabled else f'🔴 DISABLED — {killed}'}\n"
|
|
f"💰 Balance: €{acc.get('balance',0):.2f} | Equity: €{acc.get('equity',0):.2f}\n"
|
|
f"📊 Open Positions: {r.get('open_positions',0)} | "
|
|
f"Lots: {r.get('total_lots',0)} | PnL: €{r.get('unrealized_pnl',0):.2f}\n"
|
|
f"📈 Buy Level: {r.get('buy_level',0)}/{cb.get('max_levels','?')} | "
|
|
f"Sell Level: {r.get('sell_level',0)}/{cb.get('max_levels','?')}\n"
|
|
f"🎯 Cycles: {r.get('total_cycles',0)} | "
|
|
f"Journal: {j.get('wins',0)}W / {j.get('losses',0)}L\n\n"
|
|
f"🛡 Circuit Breakers:\n"
|
|
f" Max DD: {cb.get('max_dd_pct','?')}% | Daily Loss: {cb.get('max_daily_loss','?')}%\n"
|
|
f" Max Levels: {cb.get('max_levels','?')} | Max Lots: {cb.get('max_lots','?')}"
|
|
)
|
|
|
|
def dispatch(text, from_id):
|
|
if str(from_id) != TG_CHAT_ID: return
|
|
lower = text.lower().strip()
|
|
|
|
def bg(fn, *args): threading.Thread(target=fn, args=args, daemon=True).start()
|
|
|
|
if lower == "/heph_status":
|
|
bg(send_status_tg)
|
|
elif lower == "/heph_tick":
|
|
# BLOCKED — only Hermes can run ticks via CLI
|
|
tg_send(
|
|
f"🔒 *{STRATEGY}*: `/heph_tick` is reserved for Hermes only.\n"
|
|
f"Hermes runs grid ticks autonomously via CLI.\n"
|
|
f"_You will see every action here automatically._"
|
|
)
|
|
elif lower == "/heph_kill":
|
|
def run():
|
|
tg_send(f"🚨 *{STRATEGY}*: EMERGENCY KILL initiated…")
|
|
r = do_kill()
|
|
tg_send(f"🚨 *{STRATEGY}*: {'All positions closed. DISABLED.' if r.get('killed') else 'Kill failed — check logs.'}")
|
|
bg(run)
|
|
elif lower == "/heph_enable":
|
|
def run():
|
|
r = do_enable()
|
|
tg_send(f"✅ *{STRATEGY}*: {'RE-ENABLED. Hermes will resume grid ticks.' if r.get('enabled') else 'Enable failed.'}")
|
|
bg(run)
|
|
elif lower in ("/heph_help", "/heph"):
|
|
tg_send(
|
|
f"🔩 *{STRATEGY} — Strategy F (Monitor & Override)*\n\n"
|
|
f"🤖 *Hermes runs this strategy autonomously.*\n"
|
|
f"_You will receive automatic updates for every grid action._\n\n"
|
|
f"`/heph_status` — Full grid status + circuit breaker state\n"
|
|
f"`/heph_kill` — 🚨 EMERGENCY: close all positions + disable\n"
|
|
f"`/heph_enable` — Re-enable after kill\n"
|
|
f"`/heph_help` — This message\n\n"
|
|
f"⚠️ Max {CFG['grid']['max_grid_levels']} levels | "
|
|
f"{CFG['circuit_breakers']['max_equity_drawdown_pct']}% equity DD kill\n"
|
|
f"🔖 Tag: `{COMMENT}` | Pair: `{CFG['symbol']}`"
|
|
)
|
|
|
|
# ── CLI mode (Hermes calls this) ───────────────────────────────────────────────
|
|
def cli():
|
|
args = sys.argv[1:]
|
|
if not args:
|
|
print(json.dumps({"error":"Usage: hephaestus_tool.py [tick|status|kill|enable|symbols]"}))
|
|
sys.exit(1)
|
|
cmd = args[0].lower()
|
|
if cmd=="tick": print(json.dumps(do_run_tick(), indent=2, default=str))
|
|
elif cmd=="status": print(json.dumps(do_status(), indent=2, default=str))
|
|
elif cmd=="kill": print(json.dumps(do_kill(), indent=2, default=str))
|
|
elif cmd=="enable": print(json.dumps(do_enable(), indent=2, default=str))
|
|
elif cmd=="symbols": print(json.dumps({"symbol":CFG["symbol"],"comment":COMMENT,"strategy":"Grid+Martingale"},indent=2))
|
|
else: print(json.dumps({"error":f"Unknown: {cmd}"}))
|
|
|
|
# ── Bot mode ───────────────────────────────────────────────────────────────────
|
|
def bot():
|
|
log.info(f"=== {STRATEGY} Telegram Bot started ===")
|
|
tg_send(
|
|
f"🔩 *{STRATEGY} Bot Online*\n"
|
|
f"Strategy F: Grid + Martingale on `{CFG['symbol']}`\n"
|
|
f"Max Levels: {CFG['grid']['max_grid_levels']} | "
|
|
f"DD Kill: {CFG['circuit_breakers']['max_equity_drawdown_pct']}%\n\n"
|
|
f"🤖 *Hermes runs this strategy. You cannot trigger it.*\n"
|
|
f"_You will see every grid action, level open, and TP here automatically._\n\n"
|
|
f"✅ Your controls: `/heph_status` `/heph_kill` `/heph_enable`\n"
|
|
f"⚠️ Use `/heph_kill` anytime to emergency stop all grid positions."
|
|
)
|
|
offset = 0
|
|
while True:
|
|
try:
|
|
for upd in tg_updates(offset):
|
|
offset = upd["update_id"]+1
|
|
msg = upd.get("message",{})
|
|
text = msg.get("text","")
|
|
chat_id = str(msg.get("chat",{}).get("id",""))
|
|
if text.startswith("/heph"): dispatch(text, chat_id)
|
|
except Exception as e:
|
|
log.error(f"Poll error: {e}"); time.sleep(5)
|
|
time.sleep(1)
|
|
|
|
if __name__ == "__main__":
|
|
if Path(sys.argv[0]).name.startswith("hephaestus_telegram"):
|
|
bot()
|
|
else:
|
|
cli() |