β ΠΠ°Π·Π°Π΄"""
TraderMakeMoney (TMM) Journal Integration.
Auto-tags trades by strategy (WT/Scalp/Gerchik),
posts daily summaries, and syncs trade descriptions.
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(__name__)
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", "276317")) # Binance key ID in TMM
# Tag column IDs (created in TMM)
TAG_COL_STRATEGY = int(os.environ.get("TMM_TAG_COL_STRATEGY", "10")) # "Π‘ΡΡΠ°ΡΠ΅Π³ΠΈΡ"
TAG_COL_ENTRY = 1 # "ΠΡΠΈΡΠΈΠ½Ρ Π²Ρ
ΠΎΠ΄Π°" (pre-existing)
TAG_COL_EXIT = 2 # "ΠΡΠΈΡΠΈΠ½Ρ Π²ΡΡ
ΠΎΠ΄Π°" (pre-existing)
# Strategy tag names
STRATEGY_TAGS = {
"WT": "WT",
"SCALP": "Scalp",
"GERCHIK": "Gerchik",
}
# Gerchik model tags
MODEL_TAGS = {
"A": "ΠΡΠ±ΠΎΠΉ ΠΎΡ ΡΡΠΎΠ²Π½Ρ",
"B": "ΠΠΎΠΆΠ½ΡΠΉ ΠΏΡΠΎΠ±ΠΎΠΉ",
"C": "Π‘Π»ΠΎΠΆΠ½ΡΠΉ Π»ΠΎΠΆΠ½ΡΠΉ ΠΏΡΠΎΠ±ΠΎΠΉ",
"D": "ΠΡΠΎΠ±ΠΎΠΉ ΡΡΠΎΠ²Π½Ρ",
}
class TMMClient:
"""TraderMakeMoney API client for 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("TMM journal integration enabled")
else:
logger.info("TMM journal integration disabled (no API key)")
# Cache: symbol+open_time -> tmm_trade_id (avoid repeated lookups)
self._trade_cache: dict[str, int] = {}
self._pending_tags: list = []
def _get(self, path: str, params: dict = None) -> Optional[dict]:
"""GET request to TMM API."""
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} failed: {e}")
return None
def _post(self, path: str, data: dict = None) -> Optional[dict]:
"""POST request to TMM API."""
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} failed: {e}")
return None
# ββ Trade lookup ββββββββββββββββββββββββββββββββββββββββββββββ
def find_trade_by_symbol(self, symbol: str, side: str,
open_time_ms: int = 0,
search_window_ms: int = 60_000) -> Optional[int]:
"""
Find a TMM trade ID by symbol and approximate open time.
TMM auto-imports trades from Binance, so we search by symbol+time.
Returns TMM trade ID or None.
"""
cache_key = f"{symbol}_{side}_{open_time_ms}"
if cache_key in self._trade_cache:
return self._trade_cache[cache_key]
# Get recent trades sorted by open_time desc
result = self._get("/trades", params={
"page": 1,
"itemsPerPage": 20,
"sortBy": "open_time",
"sortDesc": "true",
})
if not result or "data" not in result:
return None
for trade in result["data"]:
if trade["symbol"] != symbol:
continue
t_side = trade["side"].upper()
if t_side == "LONG" and side != "BUY":
continue
if t_side == "SHORT" and side != "SELL":
continue
# If we have open_time, match within window
if open_time_ms:
t_open = trade["open_time"]
if abs(t_open - open_time_ms) <= search_window_ms:
self._trade_cache[cache_key] = trade["id"]
return trade["id"]
else:
# No time filter β return first matching symbol+side
self._trade_cache[cache_key] = trade["id"]
return trade["id"]
return None
def find_recent_trade(self, symbol: str, side: str) -> Optional[int]:
"""Find the most recent TMM trade for symbol+side (within last 5 min)."""
now_ms = int(time.time() * 1000)
return self.find_trade_by_symbol(symbol, side, now_ms, search_window_ms=300_000)
# ββ Tag management ββββββββββββββββββββββββββββββββββββββββββββ
def tag_trade(self, trade_id: int, tag_name: str, column: int = TAG_COL_STRATEGY) -> bool:
"""Add a tag to a trade. Creates tag if it doesn't exist."""
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 #{trade_id} with '{tag_name}'")
return True
return False
def tag_strategy(self, trade_id: int, strategy: str, model: str = "") -> bool:
"""Tag a trade with strategy name and optional model."""
tag_name = STRATEGY_TAGS.get(strategy, strategy)
ok = self.tag_trade(trade_id, tag_name, TAG_COL_STRATEGY)
# Add model tag for Gerchik
if model and model in MODEL_TAGS:
self.tag_trade(trade_id, MODEL_TAGS[model], TAG_COL_ENTRY)
return ok
# ββ Trade description βββββββββββββββββββββββββββββββββββββββββ
def update_description(self, trade_id: int, description: str) -> bool:
"""Update trade description/notes."""
result = self._post(f"/trades/{trade_id}/update", {
"description": description,
})
if result and result.get("status") == "success":
logger.debug(f"TMM: updated description for trade #{trade_id}")
return True
return False
# ββ Daily notes βββββββββββββββββββββββββββββββββββββββββββββββ
def post_daily_note(self, date_str: str, content: str) -> bool:
"""
Create or update a daily journal note.
date_str: "2026-03-29" format
content: markdown text for the day
"""
# Check if note already exists for this date
existing = self._get("/analyzer/notes", params={"type": 2})
note_id = None
if existing and existing.get("status") == "success":
notes = existing.get("data", {}).get("2", [])
for n in notes:
if n.get("note_at") == date_str:
note_id = n["id"]
break
payload = {
"type": 2, # 2 = day note
"note_at": date_str,
"desc": content,
}
if note_id:
payload["id"] = note_id
result = self._post("/analyzer/notes", payload)
if result and result.get("status") == "success":
logger.info(f"TMM: daily note {'updated' if note_id else 'created'} for {date_str}")
return True
return False
# ββ High-level: auto-tag on trade events ββββββββββββββββββββββ
def on_trade_opened(self, symbol: str, side: str, strategy: str,
model: str = "", signal_info: str = ""):
"""
Called when bot opens a trade. Finds it in TMM and tags it.
TMM may take a few seconds to import the trade from Binance.
"""
if not self.enabled:
return
# Delay slightly to let TMM import the trade
time.sleep(3)
trade_id = self.find_recent_trade(symbol, side)
if not trade_id:
logger.warning(f"TMM: trade not found for {symbol} {side}, will retry later")
# Store for retry
self._pending_tags.append({
"symbol": symbol, "side": side, "strategy": strategy,
"model": model, "signal_info": signal_info,
"attempts": 1, "next_retry": time.time() + 10,
})
return
self._apply_tags(trade_id, strategy, model, signal_info)
def _apply_tags(self, trade_id: int, strategy: str, model: str = "",
signal_info: str = ""):
"""Apply strategy tag + description to a trade."""
self.tag_strategy(trade_id, strategy, model)
if signal_info:
self.update_description(trade_id, signal_info)
def retry_pending_tags(self):
"""Retry tagging trades that weren't found immediately."""
if not self.enabled or not hasattr(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["strategy"],
item.get("model", ""), item.get("signal_info", ""))
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
# ββ Summaries βββββββββββββββββββββββββββββββββββββββββββββββββ
def _get_trades_for_period(self, date_from: str, date_to: str) -> list:
"""Fetch trades between two dates (YYYY-MM-DD)."""
result = self._get("/trades", params={
"api_key_id": TMM_API_KEY_ID,
"date_from": date_from,
"date_to": date_to,
"page": 1,
"itemsPerPage": 100,
"sortBy": "open_time",
"sortDesc": "true",
})
if result and "data" in result:
return result["data"]
return []
def _format_trades_summary(self, trades: list, title: str) -> str:
"""Format a list of trades into a summary message."""
if not trades:
return f"{title}\nββββββββββββββββββββ\nπ ΠΠ΅Ρ ΡΠ΄Π΅Π»ΠΎΠΊ Π·Π° ΠΏΠ΅ΡΠΈΠΎΠ΄"
# Filter only closed trades (process=0 or close_time > 0)
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 f"{title}\nββββββββββββββββββββ\nπ ΠΠ΅Ρ ΡΠ΄Π΅Π»ΠΎΠΊ Π·Π° ΠΏΠ΅ΡΠΈΠΎΠ΄"
count = len(closed)
wins = sum(1 for t in closed if float(t.get("net_profit", 0)) > 0)
losses = count - wins
total_pnl = sum(float(t.get("net_profit", 0)) for t in closed)
wr = (wins / count * 100) if count > 0 else 0
win_pnls = [float(t["net_profit"]) for t in closed if float(t.get("net_profit", 0)) > 0]
loss_pnls = [float(t["net_profit"]) for t in closed if float(t.get("net_profit", 0)) <= 0]
avg_win = sum(win_pnls) / len(win_pnls) if win_pnls else 0
avg_loss = sum(loss_pnls) / len(loss_pnls) if loss_pnls else 0
# Best and worst trade
best = max(closed, key=lambda t: float(t.get("net_profit", 0))) if closed else None
worst = min(closed, key=lambda t: float(t.get("net_profit", 0))) if closed else None
emoji = "π’" if total_pnl >= 0 else "π΄"
lines = [
f"{title}",
f"ββββββββββββββββββββ",
f"{emoji} PnL: ${total_pnl:+.2f}",
f"π ΠΠ°ΠΊΡΡΡΠΎ: {count} ({wins}W / {losses}L)",
f"π― Win Rate: {wr:.1f}%",
]
if win_pnls:
lines.append(f"π Avg Win: ${avg_win:.2f}")
if loss_pnls:
lines.append(f"π Avg Loss: ${avg_loss:.2f}")
if best:
lines.append(f"π Best: {best['symbol']} ${float(best['net_profit']):+.2f}")
if worst and count > 1:
lines.append(f"π© Worst: {worst['symbol']} ${float(worst['net_profit']):+.2f}")
if open_trades:
lines.append(f"β³ ΠΡΠΊΡΡΡΠΎ: {len(open_trades)} ΡΠ΄Π΅Π»ΠΎΠΊ")
lines.append("ββββββββββββββββββββ")
return "\n".join(lines)
def generate_today_summary(self) -> str:
"""Today's trading summary."""
today = datetime.now(VANCOUVER_TZ).strftime("%Y-%m-%d")
trades = self._get_trades_for_period(today, today)
return self._format_trades_summary(trades, f"π Π‘Π²ΠΎΠ΄ΠΊΠ° Π·Π° ΡΠ΅Π³ΠΎΠ΄Π½Ρ ({today})")
def generate_weekly_summary(self) -> str:
"""Current week summary (Mon-Sun)."""
now = datetime.now(VANCOUVER_TZ)
monday = now - timedelta(days=now.weekday())
sunday = monday + timedelta(days=6)
trades = self._get_trades_for_period(
monday.strftime("%Y-%m-%d"), sunday.strftime("%Y-%m-%d")
)
return self._format_trades_summary(
trades,
f"π ΠΠ΅Π΄Π΅Π»Ρ ({monday.strftime('%d.%m')} β {sunday.strftime('%d.%m')})"
)
def generate_monthly_summary(self) -> str:
"""Current month summary."""
now = datetime.now(VANCOUVER_TZ)
first_day = now.replace(day=1).strftime("%Y-%m-%d")
# Last day of month
if now.month == 12:
last_day = now.replace(year=now.year + 1, month=1, day=1) - timedelta(days=1)
else:
last_day = now.replace(month=now.month + 1, day=1) - timedelta(days=1)
trades = self._get_trades_for_period(first_day, last_day.strftime("%Y-%m-%d"))
month_name = now.strftime("%B %Y")
return self._format_trades_summary(trades, f"π ΠΠ΅ΡΡΡ: {month_name}")
def generate_pnl_by_strategy(self) -> str:
"""PnL breakdown by strategy tag β source of truth from Binance via TMM."""
# Get ALL trades (no date filter β full history)
result = self._get("/trades", params={
"api_key_id": TMM_API_KEY_ID,
"page": 1,
"itemsPerPage": 200,
"sortBy": "open_time",
"sortDesc": "true",
})
if not result or "data" not in result:
return "π TMM: Π½Π΅ ΡΠ΄Π°Π»ΠΎΡΡ ΠΏΠΎΠ»ΡΡΠΈΡΡ Π΄Π°Π½Π½ΡΠ΅"
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]
# Group by strategy tag
strats = {"WT": [], "Scalp": [], "Gerchik": [], "Untagged": []}
gerchik_models = {} # model_tag -> count
for t in closed:
tags = t.get("tags") or []
tag_names = [tag["name"] for tag in tags]
assigned = False
for strat_key in ["WT", "Scalp", "Gerchik"]:
if strat_key in tag_names:
strats[strat_key].append(t)
assigned = True
break
if not assigned:
strats["Untagged"].append(t)
# Count Gerchik models
for tn in tag_names:
if tn in ("ΠΡΠ±ΠΎΠΉ ΠΎΡ ΡΡΠΎΠ²Π½Ρ", "ΠΠΎΠΆΠ½ΡΠΉ ΠΏΡΠΎΠ±ΠΎΠΉ", "Π‘Π»ΠΎΠΆΠ½ΡΠΉ Π»ΠΎΠΆΠ½ΡΠΉ ΠΏΡΠΎΠ±ΠΎΠΉ", "ΠΡΠΎΠ±ΠΎΠΉ ΡΡΠΎΠ²Π½Ρ"):
gerchik_models[tn] = gerchik_models.get(tn, 0) + 1
lines = ["π° PnL Summary (TMM)", "ββββββββββββββββββββ"]
total_pnl = 0
for name, emoji in [("WT", "π"), ("Scalp", "β‘"), ("Gerchik", "π")]:
group = strats[name]
if not group:
continue
cnt = len(group)
pnl = sum(float(t.get("net_profit", 0)) for t in group)
total_pnl += pnl
wins = sum(1 for t in group if float(t.get("net_profit", 0)) > 0)
losses = cnt - wins
wr = (wins / cnt * 100) if cnt > 0 else 0
lines.append(f"\n{emoji} {name}:")
lines.append(f" Π‘Π΄Π΅Π»ΠΎΠΊ: {cnt} ({wins}W/{losses}L)")
lines.append(f" WR: {wr:.0f}%")
lines.append(f" π΅ PnL: ${pnl:+.2f}")
if name == "Gerchik" and gerchik_models:
models_str = " | ".join(f"{k}: {v}" for k, v in gerchik_models.items())
lines.append(f" π {models_str}")
# Untagged β don't include in total (pre-integration trades)
if strats["Untagged"]:
lines.append(f"\nβ ΠΠ΅Π· ΡΠ΅Π³Π°: {len(strats['Untagged'])} (Π΄ΠΎ ΠΈΠ½ΡΠ΅Π³ΡΠ°ΡΠΈΠΈ)")
if open_trades:
lines.append(f"\nβ³ ΠΡΠΊΡΡΡΠΎ: {len(open_trades)}")
lines.append(f"\nπ° TOTAL: ${total_pnl:+.2f}")
lines.append("ββββββββββββββββββββ")
return "\n".join(lines)
_pending_tags: list = []