"""
TraderMakeMoney (TMM) Journal Integration for Squeeze-VWAP Bot.
Auto-tags trades with "SqzVWAP" strategy tag on Bybit.
TMM API key ID: 276474 (bybit-tiger, Bybit Futures).
API docs: https://tradermake.money/api/v2/docs/v2
"""
import logging
import os
import time
from datetime import datetime, timezone, timedelta
from typing import Optional
import requests
logger = logging.getLogger("tmm")
VANCOUVER_TZ = timezone(timedelta(hours=-7))
TMM_API_KEY = os.environ.get("TMM_API_KEY", "")
TMM_BASE_URL = "https://tradermake.money/api/v2"
TMM_API_KEY_ID = int(os.environ.get("TMM_API_KEY_ID", "276474")) # Bybit key ID in TMM
# Tag column IDs
TAG_COL_STRATEGY = int(os.environ.get("TMM_TAG_COL_STRATEGY", "10")) # "Π‘ΡΡΠ°ΡΠ΅Π³ΠΈΡ"
STRATEGY_TAG = "SqzVWAP"
class TMMClient:
"""TraderMakeMoney API client for Squeeze-VWAP journal integration."""
def __init__(self):
self.api_key = TMM_API_KEY
self.base_url = TMM_BASE_URL
self.session = requests.Session()
self.session.headers.update({
"API-KEY": self.api_key,
"Content-Type": "application/json",
})
self.enabled = bool(self.api_key)
if self.enabled:
logger.info(f"TMM enabled (Bybit key #{TMM_API_KEY_ID})")
else:
logger.info("TMM disabled (no API key)")
self._pending_tags: list = []
def _get(self, path: str, params: dict = None) -> Optional[dict]:
if not self.enabled:
return None
try:
resp = self.session.get(f"{self.base_url}{path}", params=params, timeout=10)
resp.raise_for_status()
return resp.json()
except Exception as e:
logger.error(f"TMM GET {path}: {e}")
return None
def _post(self, path: str, data: dict = None) -> Optional[dict]:
if not self.enabled:
return None
try:
resp = self.session.post(f"{self.base_url}{path}", json=data, timeout=10)
resp.raise_for_status()
return resp.json()
except Exception as e:
logger.error(f"TMM POST {path}: {e}")
return None
# ββ Trade lookup ββββββββββββββββββββββββββββββββββββββββββ
def find_recent_trade(self, symbol: str, side: str) -> Optional[int]:
"""
Find the most recent TMM trade for symbol+side that hasn't been tagged yet.
TMM auto-imports trades from Bybit β search by symbol+side.
No cache β always fresh lookup to avoid returning stale trade IDs
when same symbol opens multiple times.
"""
result = self._get("/trades", params={
"api_key_id": TMM_API_KEY_ID,
"page": 1,
"itemsPerPage": 20,
"sortBy": "open_time",
"sortDesc": "true",
})
if not result or "data" not in result:
return None
now_ms = int(time.time() * 1000)
for trade in result["data"]:
if trade["symbol"] != symbol:
continue
t_side = trade["side"].upper()
# Bybit: BUY=LONG, SELL=SHORT
if side.upper() == "BUY" and t_side != "LONG":
continue
if side.upper() == "SELL" and t_side != "SHORT":
continue
# Skip already tagged trades (avoid re-tagging wrong trade)
existing_tags = [tag["name"] for tag in (trade.get("tags") or [])]
if STRATEGY_TAG in existing_tags:
continue
# Within last 10 min (was 5 min β TMM import can be slow)
t_open = trade.get("open_time", 0)
if abs(now_ms - t_open) <= 600_000:
return trade["id"]
return None
# ββ Tagging βββββββββββββββββββββββββββββββββββββββββββββββ
def tag_trade(self, trade_id: int, tag_name: str, column: int = TAG_COL_STRATEGY) -> bool:
result = self._post(f"/trades/{trade_id}/tags", {
"tags": [{"name": tag_name, "column": column}],
})
if result and result.get("status") == "success":
logger.info(f"TMM: tagged #{trade_id} β '{tag_name}'")
return True
logger.warning(f"TMM: tag failed #{trade_id}")
return False
def update_description(self, trade_id: int, description: str) -> bool:
result = self._post(f"/trades/{trade_id}/update", {
"description": description,
})
if result and result.get("status") == "success":
logger.debug(f"TMM: desc updated #{trade_id}")
return True
return False
# ββ High-level: auto-tag on trade open ββββββββββββββββββββ
def on_trade_opened(self, symbol: str, side: str, score: int,
z_score: float, reasons: list):
"""
Called after bot opens a trade.
Finds it in TMM (Bybit auto-import) and tags it.
"""
if not self.enabled:
return
# Wait for TMM to import from Bybit (5s β TMM can be slow)
time.sleep(5)
order_side = "BUY" if side == "LONG" else "SELL"
trade_id = self.find_recent_trade(symbol, order_side)
if not trade_id:
logger.warning(f"TMM: trade not found {symbol} {side}, will retry")
self._pending_tags.append({
"symbol": symbol, "side": order_side,
"score": score, "z_score": z_score,
"reasons": reasons,
"attempts": 1, "next_retry": time.time() + 15,
})
return
self._apply_tags(trade_id, score, z_score, reasons)
def _apply_tags(self, trade_id: int, score: int, z_score: float, reasons: list):
"""Apply strategy tag + description."""
self.tag_trade(trade_id, STRATEGY_TAG)
desc = (
f"Squeeze-VWAP Bot\n"
f"Score: {score}/5 | Z-VWAP: {z_score:+.2f}\n"
f"Reasons: {' | '.join(reasons[:4])}"
)
self.update_description(trade_id, desc)
def retry_pending_tags(self):
"""Retry tagging trades that weren't found immediately."""
if not self.enabled or not self._pending_tags:
return
now = time.time()
remaining = []
for item in self._pending_tags:
if now < item["next_retry"]:
remaining.append(item)
continue
trade_id = self.find_recent_trade(item["symbol"], item["side"])
if trade_id:
self._apply_tags(trade_id, item["score"],
item["z_score"], item["reasons"])
elif item["attempts"] < 10:
item["attempts"] += 1
item["next_retry"] = now + 20
remaining.append(item)
else:
logger.warning(f"TMM: gave up tagging {item['symbol']} after 10 attempts")
self._pending_tags = remaining
# ββ Summaries βββββββββββββββββββββββββββββββββββββββββββββ
def generate_summary(self) -> str:
"""PnL summary from TMM (Bybit trades only)."""
result = self._get("/trades", params={
"api_key_id": TMM_API_KEY_ID,
"page": 1,
"itemsPerPage": 100,
"sortBy": "open_time",
"sortDesc": "true",
})
if not result or "data" not in result:
return "π TMM: no data"
trades = result["data"]
closed = [t for t in trades if int(t.get("close_time", 0)) > 0]
open_trades = [t for t in trades if int(t.get("close_time", 0)) == 0]
if not closed and not open_trades:
return "π TMM: no trades yet"
# Filter SqzVWAP tagged
tagged = []
untagged = []
for t in closed:
tags = [tag["name"] for tag in (t.get("tags") or [])]
if STRATEGY_TAG in tags:
tagged.append(t)
else:
untagged.append(t)
lines = ["π *TMM Squeeze\\-VWAP Summary*", "ββββββββββββββββββββ"]
for label, group in [("SqzVWAP", tagged), ("Other", untagged)]:
if not group:
continue
cnt = len(group)
pnl = sum(float(t.get("net_profit", 0)) for t in group)
wins = sum(1 for t in group if float(t.get("net_profit", 0)) > 0)
wr = (wins / cnt * 100) if cnt > 0 else 0
emoji = "π’" if pnl >= 0 else "π΄"
lines.append(f"{emoji} {label}: {cnt} trades, WR {wr:.0f}%, ${pnl:+.2f}")
if open_trades:
lines.append(f"β³ Open: {len(open_trades)}")
total = sum(float(t.get("net_profit", 0)) for t in closed)
lines.append(f"\nπ° Total: ${total:+.2f}")
return "\n".join(lines)