← Назад"""
WT Bot v3 — TraderMakeMoney Journal Integration
=================================================
Auto-tags trades in TMM with strategy, entry reasons.
Pulls stats from TMM (source of truth).
API: https://tradermake.money/api/v2/docs/v2
Gotchas from v2:
- notes field is `desc` not `content`
- date field is `note_at`
- TMM may take 3-15s to import trade from Binance → retry 5x
"""
import logging
import os
import time
from datetime import datetime, timezone, timedelta
from typing import Optional
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 column IDs
TAG_COL_STRATEGY = int(os.environ.get("TMM_TAG_COL_STRATEGY", "10"))
TAG_COL_ENTRY = 1 # "Причины входа"
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._trade_cache = {}
if self.enabled:
logger.info("TMM integration enabled")
else:
logger.warning("TMM disabled (no API key)")
# ============================================================
# HTTP
# ============================================================
def _get(self, path, params=None):
if not self.enabled:
return None
try:
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:
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)."""
cache_key = f"{symbol}_{side}"
if cache_key in self._trade_cache:
return self._trade_cache[cache_key]
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
# Within 5 min
if abs(trade["open_time"] - now_ms) <= 300_000:
self._trade_cache[cache_key] = trade["id"]
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}'")
return True
return False
def update_description(self, trade_id, description):
"""Update trade notes/description."""
result = self._post(f"/trades/{trade_id}/update", {
"description": description,
})
if result and result.get("status") == "success":
logger.debug(f"TMM: description updated #{trade_id}")
return True
return False
# ============================================================
# ON TRADE OPENED — auto-tag with retry
# ============================================================
def on_trade_opened(self, symbol, side, wt1, wt2, zone, atr_pct=0, ema200=0, price=0):
"""
Called when bot opens a trade.
Tags: strategy "WT_v3", entry reason.
Description: WT values, ATR, EMA200, zone.
"""
if not self.enabled:
return
# Build entry reason
zone_str = "Oversold → Long" if zone == 1 else "Overbought → Short"
entry_reason = f"WT Cross {zone_str}"
# Description with all context
description = (
f"S3_EMA_Filter 5m | {zone_str}\n"
f"WT1={wt1:.1f} WT2={wt2:.1f}\n"
f"EMA200={'above' if zone == 1 else 'below'}\n"
f"ATR={atr_pct:.1f}%"
)
# НЕ блокируем — сразу в очередь с задержкой 3 сек
order_side = "BUY" if side == "LONG" else "SELL"
logger.info(f"TMM: queuing tag for {symbol} {order_side}")
self._pending_tags.append({
"symbol": symbol, "side": order_side,
"entry_reason": entry_reason, "description": description,
"attempts": 0, "next_retry": time.time() + 3,
})
def _apply_tags(self, trade_id, entry_reason, description):
"""Apply strategy tag + entry reason + description."""
# Strategy tag
self.tag_trade(trade_id, "WT_v3", TAG_COL_STRATEGY)
# Entry reason
self.tag_trade(trade_id, entry_reason, TAG_COL_ENTRY)
# Description
self.update_description(trade_id, description)
# Чистим кэш чтоб повторные сделки на том же символе не конфликтовали
stale_keys = [k for k, v in self._trade_cache.items() if v == trade_id]
for k in stale_keys:
del self._trade_cache[k]
def retry_pending_tags(self):
"""Retry tagging trades that weren't found immediately. Call from main loop."""
if 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["entry_reason"], item["description"])
elif item["attempts"] < 5:
item["attempts"] += 1
item["next_retry"] = now + 15
remaining.append(item)
else:
logger.warning(f"TMM: gave up tagging {item['symbol']} after 5 attempts")
self._pending_tags = remaining
# ============================================================
# STATS FROM TMM (source of truth)
# ============================================================
def get_today_summary(self):
"""Today's summary from TMM."""
today = datetime.now(VANCOUVER_TZ).strftime("%Y-%m-%d")
trades = self._get_trades(today, today)
return self._format_summary(trades, f"📊 Сегодня ({today})")
def get_weekly_summary(self):
"""Current week summary."""
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"📊 Неделя ({monday.strftime('%d.%m')}→{sunday.strftime('%d.%m')})")
def get_total_summary(self):
"""All-time summary."""
trades = self._get_trades("2020-01-01", "2099-12-31")
return self._format_summary(trades, "📊 Всего (WT_v3)")
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📭 Нет сделок"
closed = [t for t in trades if int(t.get("close_time", 0)) > 0]
open_t = [t for t in trades if int(t.get("close_time", 0)) == 0]
if not closed:
msg = f"{title}\n📭 Нет закрытых сделок"
if open_t:
msg += f"\n⏳ Открыто: {len(open_t)}"
return msg
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)
losses = len(closed) - wins
wr = wins / len(closed) * 100 if closed else 0
emoji = "🟢" if total_pnl >= 0 else "🔴"
lines = [
title,
f"{emoji} PnL: ${total_pnl:+.2f}",
f"📈 Сделок: {len(closed)} ({wins}W/{losses}L)",
f"🎯 WR: {wr:.0f}%",
]
if closed:
best = max(closed, key=lambda t: float(t.get("net_profit", 0)))
worst = min(closed, key=lambda t: float(t.get("net_profit", 0)))
lines.append(f"🏆 Best: {best['symbol']} ${float(best['net_profit']):+.2f}")
if len(closed) > 1:
lines.append(f"💩 Worst: {worst['symbol']} ${float(worst['net_profit']):+.2f}")
if open_t:
lines.append(f"⏳ Открыто: {len(open_t)}")
return "\n".join(lines)