โ ะะฐะทะฐะด"""
Grid Bot โ TraderMakeMoney Journal Integration
=================================================
Auto-tags grid trades in TMM.
Multi-tags: Column 10 = Grid.v1-{spacing} (strategy), Column 1 = coin name.
IMPORTANT: Only tag on ROUND TRIPS (not every fill).
Grids generate tons of fills โ tagging each one = 429 rate limit.
Rate limit: min 3 sec between API calls, dedup already-tagged trades.
"""
import logging
import os
import time
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
import requests
logger = logging.getLogger("tmm")
VANCOUVER_TZ = ZoneInfo("America/Vancouver")
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", "276317"))
TAG_COL_STRATEGY = int(os.environ.get("TMM_TAG_COL_STRATEGY", "10"))
TAG_COL_ENTRY = 1
# Rate limit: min seconds between TMM API calls
MIN_API_INTERVAL = 3.0
class TMMClient:
def __init__(self):
self.api_key = TMM_API_KEY
self.session = requests.Session()
self.session.headers.update({
"API-KEY": self.api_key,
"Content-Type": "application/json",
})
self.enabled = bool(self.api_key)
self._pending_tags = []
self._tagged_trade_ids = set() # dedup: don't re-tag
self._last_api_call = 0.0 # rate limiter
self._entry_indicators = {} # screener data at session start
if self.enabled:
logger.info("TMM integration enabled")
else:
logger.warning("TMM disabled (no API key)")
# ============================================================
# HTTP (with rate limiting)
# ============================================================
def _rate_wait(self):
"""Wait if needed to respect TMM rate limit."""
now = time.time()
elapsed = now - self._last_api_call
if elapsed < MIN_API_INTERVAL:
time.sleep(MIN_API_INTERVAL - elapsed)
self._last_api_call = time.time()
def _get(self, path, params=None):
if not self.enabled:
return None
try:
self._rate_wait()
resp = self.session.get(f"{TMM_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, data=None):
if not self.enabled:
return None
try:
self._rate_wait()
resp = self.session.post(f"{TMM_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
# ============================================================
# FIND TRADE
# ============================================================
def find_recent_trade(self, symbol, side):
"""Find most recent TMM trade for symbol+side (within 5 min)."""
result = self._get("/trades", params={
"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()
if (t_side == "LONG" and side != "BUY") or (t_side == "SHORT" and side != "SELL"):
continue
trade_id = trade["id"]
# Skip already tagged
if trade_id in self._tagged_trade_ids:
continue
if abs(trade["open_time"] - now_ms) <= 300_000:
return trade_id
return None
# ============================================================
# TAG & DESCRIBE
# ============================================================
def tag_trade(self, trade_id, tag_name, column=TAG_COL_STRATEGY):
"""Add tag to trade."""
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}' (col={column})")
return True
return False
def update_description(self, trade_id, description):
"""Update trade notes."""
result = self._post(f"/trades/{trade_id}/update", {
"description": description,
})
if result and result.get("status") == "success":
logger.info(f"TMM: description updated #{trade_id}")
return True
return False
# ============================================================
# GRID EVENTS
# ============================================================
def on_grid_started(self, symbol, score_data=None):
"""Called when grid starts on a symbol. Saves entry indicators for descriptions."""
self._entry_indicators = {}
if score_data:
self._entry_indicators = {
"chop": score_data.get("chop", 0),
"adx": score_data.get("adx", 0),
"bb_width": score_data.get("bb_width", 0),
"micro_vol": score_data.get("micro_vol", 0),
"natr": score_data.get("natr", 0),
"atr_spacing": score_data.get("atr_spacing", 0),
"score": score_data.get("score", 0),
"volume_m": round(score_data.get("volume", 0) / 1e6, 0),
}
logger.info(f"TMM: grid started on {symbol}, entry: {self._entry_indicators}")
def on_round_trip(self, symbol, rt_data):
"""
Called on completed round-trip.
Tags BOTH sides: Grid.v1-{spacing} (col 10) + COIN (col 1).
Only source of TMM tags โ NOT on every fill.
"""
if not self.enabled:
return
from src.config import TMM_STRATEGY_TAG, GRID_SPACING_PCT
coin_name = symbol.replace("USDT", "").replace("1000", "")
combined_tag = f"Grid.{coin_name}"
# Entry indicators from screener
ei = getattr(self, '_entry_indicators', {})
entry_line = ""
if ei:
entry_line = (
f"\n--- Entry Indicators ---\n"
f"Score: {ei.get('score', 0):.0f} | "
f"CHOP: {ei.get('chop', 0):.1f} | "
f"ADX: {ei.get('adx', 0):.1f}\n"
f"BB: {ei.get('bb_width', 0):.2f}% | "
f"MV: {ei.get('micro_vol', 0):.0f} | "
f"NATR: {ei.get('natr', 0):.2f}%\n"
f"Vol: ${ei.get('volume_m', 0):.0f}M | "
f"Spacing: {ei.get('atr_spacing', 0):.3f}%"
)
description = (
f"Grid RT #{rt_data.get('rt_num', '?')}\n"
f"Buy: ${rt_data.get('buy_price', 0):.6f}\n"
f"Sell: ${rt_data.get('sell_price', 0):.6f}\n"
f"PnL: ${rt_data.get('pnl', 0):.6f}\n"
f"Fee: ${rt_data.get('fee', 0):.6f}"
f"{entry_line}"
)
# Queue tag for BUY side only โ TMM merges grid fills into one LONG trade
self._pending_tags.append({
"symbol": symbol,
"side": "BUY",
"combined_tag": combined_tag,
"description": description,
"attempts": 0,
"next_retry": time.time() + 8, # wait 8s for TMM to import (was 5)
})
logger.info(
f"TMM: RT #{rt_data.get('rt_num', '?')} {symbol} "
f"PnL: ${rt_data.get('pnl', 0):.6f} โ queued for tagging"
)
def _apply_tags(self, trade_id, item):
"""Apply combined tag (Grid.COIN) + description. One tag = fewer API calls."""
self.tag_trade(trade_id, item["combined_tag"], TAG_COL_ENTRY)
self.update_description(trade_id, item["description"])
# Mark as tagged (dedup)
self._tagged_trade_ids.add(trade_id)
# Keep set bounded (last 500)
if len(self._tagged_trade_ids) > 500:
self._tagged_trade_ids = set(list(self._tagged_trade_ids)[-300:])
def retry_pending_tags(self):
"""Retry tagging trades. Called from main loop. Process max 1 per tick."""
if not self._pending_tags:
return
now = time.time()
remaining = []
# Process only ONE pending tag per tick to avoid API spam
processed_one = False
for item in self._pending_tags:
if processed_one or 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)
processed_one = True
elif item["attempts"] < 8:
item["attempts"] += 1
item["next_retry"] = now + 20 # retry every 20s, up to 8 attempts (~3min)
remaining.append(item)
processed_one = True # still counts as API call
else:
logger.warning(f"TMM: gave up tagging {item['symbol']} {item['side']} after 8 attempts")
self._pending_tags = remaining
# ============================================================
# STATS
# ============================================================
def get_today_summary(self):
today = datetime.now(VANCOUVER_TZ).strftime("%Y-%m-%d")
trades = self._get_trades(today, today)
return self._format_summary(trades, f"๐ Today ({today})")
def get_weekly_summary(self):
now = datetime.now(VANCOUVER_TZ)
monday = now - timedelta(days=now.weekday())
sunday = monday + timedelta(days=6)
trades = self._get_trades(monday.strftime("%Y-%m-%d"), sunday.strftime("%Y-%m-%d"))
return self._format_summary(trades, f"๐ Week ({monday.strftime('%d.%m')}โ{sunday.strftime('%d.%m')})")
def _get_trades(self, date_from, date_to):
result = self._get("/trades", params={
"api_key_id": TMM_API_KEY_ID,
"date_from": date_from, "date_to": date_to,
"page": 1, "itemsPerPage": 200,
"sortBy": "open_time", "sortDesc": "true",
})
if result and "data" in result:
return result["data"]
return []
def _format_summary(self, trades, title):
if not trades:
return f"{title}\n๐ญ No trades"
closed = [t for t in trades if int(t.get("close_time", 0)) > 0]
if not closed:
open_count = len([t for t in trades if int(t.get("close_time", 0)) == 0])
return f"{title}\n๐ญ No closed trades\nโณ Open: {open_count}"
total_pnl = sum(float(t.get("net_profit", 0)) for t in closed)
wins = sum(1 for t in closed if float(t.get("net_profit", 0)) > 0)
wr = wins / len(closed) * 100
emoji = "๐ข" if total_pnl >= 0 else "๐ด"
return (
f"{title}\n"
f"{emoji} PnL: ${total_pnl:+.2f}\n"
f"๐ Trades: {len(closed)} ({wins}W/{len(closed)-wins}L)\n"
f"๐ฏ WR: {wr:.0f}%"
)