I log all activity, including a session P&L, ongoing P&L etc and alert to Telegram so I can keep an eye on results remotely. the P&L data is logged to a csv so it will resume the bank and ongoing totals each day
I've seen some discussions on the inclusion of the customerStrategyRef & customerOrderRef from BF in the new beta so individual strategy results can be tracked but it looks like that's not coming any time soon.
Given my code tracks by stake placed, matched status for backs and lays etc I'm using this in it's place and it's working very well for me. With a little editing and staking with slightly different values it should easily be possible to track and log results for different markets, and maintain separate P&L for each
So, I've stripped out all my strategy code from the script and am posting this up as a generic pnl tracker and BA loader. I only use it for horses myself but it should work on any other markets if you make a few edits.
For each runner it will track backs and lays , check if they are matched, calc the P&L per runner, log to the console and the output files. I've only been using it to track straight lays in play myself but it should support trading as well. It will also subtract 2% commission (set in the top config section)
You'll need to make a few edits to load your own coupons, apply your own baf, import Dallas' Log Winner baf to BA (that I've attached) to track race results (winner\loser). Posting up into any AI is probably the fastest way to summarise what it's doing and how to edit it for your own use. You'll also need to create your own Telegram bot and edit the config section to get your own alerts
I've quickly edited this to remove all my own strategy code but it should work as designed
I'm posting this unsupported as I don't have the time to maintain it as it's a generic version but hopefully it might help someone, and save a good few hours coding or battling with AI
Code: Select all
#
# BetAngel_BAF_PnL_Monitor.py
#
# Generic P&L monitor for Bet Angel.
# - This script does NOT place any bets.
# - It is designed to work WITH an automation file (BAF)
# that places bets and logs the result to a Stored Value.
#
# --- FEATURES ---
# - Applies a Guardian coupon ("Todays_Horses") on startup.
# - Applies an automation rules file ("Log Winner") to all markets.
# - Monitors all markets post-off.
# - Discovers all matched Back/Lay bets in those markets.
# - Reads a stored value ("result") for the selection to get the outcome.
# - Calculates P&L (inc. commission) for all discovered bets.
# - Logs all P&L to a CSV file (generic_pnl_log.csv).
# - Maintains a virtual bank balance across sessions.
# - Sends Telegram alerts for startup and per-race P&L.
#
# --- HOW TO USE ---
# 1. Your automation file ("Log Winner.baf") MUST write the result
# for a selection to a Stored Value named "result" (e.g., "WIN" or "LOSE").
# 2. Configure your Telegram Token and Chat ID below.
# 3. Run the script. It will monitor in the background.
#
# -----------------------------------------------------------------------------
import os
import sys
import json
import time
import logging
import traceback
import csv
from datetime import datetime, timedelta, timezone
from typing import Dict, Any, List, Optional, Tuple, Set
import requests
import pandas as pd # Used only for final bet log dump
# ------------------------------ CONFIG ----------------------------------
BASE_URL = os.environ.get("BETANGEL_BASE_URL", "http://localhost:9000/api/")
# --- Log Paths ---
LOG_PATH = os.environ.get("LOG_PATH", r"C:\Temp\generic_baf_monitor.log")
PNL_LOG_PATH = os.environ.get("PNL_LOG_PATH", r"C:\Temp\generic_pnl_log.csv")
BET_LOG_DIR = os.environ.get("BET_LOG_DIR", r"C:\Temp") # For session CSV dump
# --- Bot Settings ---
POLL_INTERVAL_SEC = 0.5
PERIODIC_DISCOVERY_SEC = 30
POST_OFF_CLOSE_WAIT_MIN = 45 # How long to wait after off-time to close a market
COMMISSION_RATE = float(os.environ.get("COMMISSION_RATE", 0.02))
STARTING_BANK = 100.0
# How long after scheduled-off to start scanning for matched bets
DISCOVER_BETS_AFTER_SEC = 60
# How often to re-scan for *new* matched bets (e.g., in-play bets)
REDISCOVER_BETS_FREQ_SEC = 30
# --- TELEGRAM CONFIG ---
# Replace with your values
TELEGRAM_BOT_TOKEN = os.environ.get("TELEGRAM_BOT_TOKEN", "YOUR_BOT_TOKEN_HERE") # <-- EDIT THIS
TELEGRAM_CHAT_ID = os.environ.get("TELEGRAM_CHAT_ID", "YOUR_CHAT_ID_HERE") # <-- EDIT THIS
# ------------------------------
# ------------------------------ LOGGING ----------------------------------
logger = logging.getLogger("baf_pnl_monitor")
logger.setLevel(logging.INFO)
_fh = logging.FileHandler(LOG_PATH, mode="a", encoding="utf-8")
_fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
_fh.setFormatter(_fmt)
logger.addHandler(_fh)
logger.addHandler(logging.StreamHandler(sys.stdout))
# ------------------------------ API WRAPPER ------------------------------
class BetAngelAPI:
def __init__(self, base_url: str = BASE_URL):
if not base_url.endswith("/"):
base_url += "/"
self.base_url = base_url
self.headers = {"Content-Type": "application/json"}
def apply_coupon(self):
url = f"{self.base_url}guardian/v1.0/applyCoupon"
payload = {"couponName": "Todays_Horses", "clearOption": "CLEAR_GUARDIAN_AND_WATCH_LIST", "watchListNumber": 1}
try:
r = requests.post(url, headers=self.headers, json=payload)
r.raise_for_status()
logger.info(f"Coupon applied: {r.json()}")
except requests.RequestException as e:
logger.error(f"Error applying coupon: {e}")
def apply_rules(self, market_ids):
url = f"{self.base_url}guardian/v1.0/applyRules"
payload = {
"rulesFileName": "Log Winner", # This BAF file MUST log the result
"marketsFilter": {"filter": "SPECIFIED_IDS", "ids": market_ids},
"guardianRulesColumn": 1
}
try:
r = requests.post(url, headers=self.headers, json=payload)
r.raise_for_status()
logger.info(f"Rules applied: {r.json()}")
except requests.RequestException as e:
logger.error(f"Error applying rules: {e}")
def get_markets(self, data_required: Optional[List[str]] = None) -> Dict[str, Any]:
if data_required is None:
data_required = [
"ID","NAME","MARKET_START_TIME","EVENT_ID","EVENT_TYPE_ID",
"MARKET_TYPE","SELECTION_IDS","SELECTION_NAMES","START_TIME"
]
url = f"{self.base_url}markets/v1.0/getMarkets"
r = requests.post(url, headers=self.headers, json={"dataRequired": data_required}, timeout=20)
r.raise_for_status()
return r.json()
def get_market_bets(self, market_id: str, option: str = "ALL") -> Dict[str, Any]:
url = f"{self.base_url}markets/v1.0/getMarketBets"
r = requests.post(url, headers=self.headers, json={"marketId": market_id, "filter": {"option": option}}, timeout=20)
r.raise_for_status()
return r.json()
def get_stored_values_for_selection(self, selection_id: int) -> List[Dict[str, Any]]:
url = f"{self.base_url}automation/v1.0/getStoredValues"
payload = {
"marketsFilter": {"filter":"ALL"},
"selectionsFilter": {"filter":"SPECIFIED_IDS","ids":[str(selection_id)]},
"storedValueFilterSelectionLevel": {"storedValueFilter":"ALL"}
}
r = requests.post(url, headers=self.headers, json=payload, timeout=20)
r.raise_for_status()
result = r.json().get("result", {})
out = []
for m in result.get("markets", []):
for sel in m.get("selections", []):
if str(sel.get("id")) == str(selection_id):
out.extend(sel.get("sharedValues", []))
return out
# ------------------------------ UTILS ------------------------------------
def safe_float(x: Any, default: float = 0.0) -> float:
try: return float(x)
except Exception: return default
def _parse_start(start_ts: Any) -> Optional[datetime]:
try:
if isinstance(start_ts, (int, float)):
return datetime.fromtimestamp(start_ts/1000.0, tz=timezone.utc)
s = str(start_ts).strip().replace("Z","+00:00")
dt = datetime.fromisoformat(s)
return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
except Exception:
return None
# ------------------------------ CORE MONITOR ---------------------------------
class BafPnLMonitor:
def __init__(self, api: BetAngelAPI):
self.api = api
# Bookkeeping
self.market_starts: Dict[str, datetime] = {}
self.open_markets: Dict[str, Dict[str,Any]] = {} # { market_id -> meta }
self.bet_log: List[Dict[str,Any]] = [] # For final CSV dump
# --- P&L TRACKING ---
self.running_pnl: float = 0.0
self.virtual_bank: float = STARTING_BANK
self.pnl_log_path: str = PNL_LOG_PATH
self._initialize_pnl_log()
# --------------------
# --- TELEGRAM CONFIG ---
self.telegram_bot_token = TELEGRAM_BOT_TOKEN
self.telegram_chat_id = TELEGRAM_CHAT_ID
if not self.telegram_bot_token or "YOUR_BOT" in self.telegram_bot_token:
logger.warning("TELEGRAM_BOT_TOKEN is not set. Alerts will be disabled.")
self.telegram_bot_token = None
if not self.telegram_chat_id or "YOUR_CHAT" in self.telegram_chat_id:
logger.warning("TELEGRAM_CHAT_ID is not set. Alerts will be disabled.")
self.telegram_chat_id = None
# -----------------------
logger.info("Starting BAF P&L Monitor")
# --- SEND STARTUP ALERT ---
self._send_startup_alert()
# --------------------------
# --- START P&L HELPER METHODS ---
def _initialize_pnl_log(self):
"""Initializes the P&L log, creating it if it doesn't exist and loading the last bank value."""
write_header = not os.path.exists(self.pnl_log_path)
if not write_header:
try:
# Read the last line to get the last bank value
df = pd.read_csv(self.pnl_log_path)
if not df.empty:
last_bank = df['Virtual_Bank'].iloc[-1]
self.virtual_bank = float(last_bank)
except Exception as e:
logger.warning(f"Could not read existing P&L log '{self.pnl_log_path}'. Resetting bank. Error: {e}")
self.virtual_bank = STARTING_BANK
write_header = True # Force rewrite if file is corrupt
try:
with open(self.pnl_log_path, 'a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
if write_header:
writer.writerow(['Timestamp', 'Market_ID', 'Selection_ID', 'Bet_Pnl', 'Running_Session_Profit', 'Virtual_Bank'])
# Log a session start
writer.writerow([
datetime.utcnow().isoformat(),
'SESSION_START',
'N/A',
0.0,
self.running_pnl, # Will be 0.0 on a new session
self.virtual_bank
])
logger.info(f"P&L log initialized. Starting virtual bank: {self.virtual_bank:.2f}")
except Exception as e:
logger.error(f"Failed to initialize P&L log at '{self.pnl_log_path}': {e}")
def _update_pnl_csv(self, market_id: str, selection_id: int, bet_pnl: float):
"""Appends a single P&L record to the CSV log."""
row = [
datetime.utcnow().isoformat(),
market_id,
selection_id,
bet_pnl,
self.running_pnl,
self.virtual_bank
]
try:
with open(self.pnl_log_path, 'a', newline='', encoding='utf-8') as f:
writer = csv.writer(f)
writer.writerow(row)
except Exception as e:
logger.error(f"Failed to append to PNL CSV log: {e}")
# --- END P&L HELPER METHODS ---
# --- START TELEGRAM HELPER METHODS ---
def _send_telegram_alert(self, message_text: str):
"""Sends a message to the configured Telegram chat."""
if not self.telegram_bot_token or not self.telegram_chat_id:
logger.debug("Telegram alert skipped: Token or Chat ID not set.")
return # Silently skip if config is missing
url = f"https://api.telegram.org/bot{self.telegram_bot_token}/sendMessage"
payload = {
"chat_id": self.telegram_chat_id,
"text": message_text,
"parse_mode": "Markdown"
}
try:
response = requests.post(url, json=payload, timeout=5)
if response.status_code != 200:
logger.warning(f"Failed to send Telegram alert. Status: {response.status_code}, Response: {response.text}")
except Exception as e:
logger.error(f"Error sending Telegram alert: {e}")
def _send_startup_alert(self):
"""Sends the initial startup message to Telegram."""
today = datetime.utcnow().strftime('%Y-%m-%d')
message = (
f"?? **BAF P&L Monitor Started**\n"
f"--------------------\n"
f"**Date:** {today}\n"
f"**Starting Bank:** £{self.virtual_bank:.2f}\n"
f"**Session P&L:** £{self.running_pnl:.2f}"
)
self._send_telegram_alert(message)
# --- END TELEGRAM HELPER METHODS ---
def discover_markets(self) -> None:
"""Finds upcoming markets and stores their metadata."""
try:
js = self.api.get_markets()
except Exception as e:
logger.warning(f"discover_markets request failed: {e}"); return
markets = js.get("result", {}).get("markets", [])
now = datetime.utcnow().replace(tzinfo=timezone.utc)
enrolled = 0
for m in markets:
mid = str(m.get("id") or m.get("marketId"))
start_ts = (m.get("startTime") or m.get("marketStartTime") or m.get("MARKET_START_TIME"))
start = _parse_start(start_ts)
if start is None: continue
self.market_starts[mid] = start
minutes_to_off = (start - now).total_seconds()/60.0
# Watch markets from 20 mins before off until 5 mins after
if -5 <= minutes_to_off <= 20:
if mid not in self.open_markets:
self.open_markets[mid] = {
"name": m.get("name"),
"start": start,
"bets_to_track": [],
"processed_bet_refs": set(),
"last_bet_discovery_time": None
}
enrolled += 1
if enrolled:
logger.info(f"discover_markets: enrolled={enrolled} new markets. total_open={len(self.open_markets)}")
def _read_result_flag(self, selection_id: int) -> Optional[str]:
"""Reads win/loss result from Bet Angel stored values."""
try:
values = self.api.get_stored_values_for_selection(selection_id)
for v in values:
key_name = (v.get("n") or "").lower()
if key_name in ("result", "winner", "win"):
value = v.get("t") or v.get("v")
return str(value)
except Exception as e:
logger.warning(f"getStoredValues error for sel {selection_id}: {e}")
return None
def _discover_new_matched_bets(self, market_id: str, meta: Dict[str, Any]):
"""
Scans market for matched bets and adds any new ones to the 'bets_to_track' list.
"""
new_bets_found = 0
try:
js = self.api.get_market_bets(market_id, option="ALL")
matched_bets = js.get("result", {}).get("matchedBets", [])
if not matched_bets:
return # No matched bets in market
bets_to_track = meta.setdefault("bets_to_track", [])
processed_refs = meta.setdefault("processed_bet_refs", set())
for b in matched_bets:
bet_ref = b.get("betRef") or b.get("betId")
if not bet_ref or str(bet_ref) in processed_refs:
continue # Skip if no ID or already processed
# This is a new bet we haven't seen before
try:
bet_data = {
"selectionId": int(b["selectionId"]),
"betRef": str(bet_ref),
"price": safe_float(b.get("price")),
"stake": safe_float(b.get("stake")),
"type": b.get("type", "").upper(),
"settled": False,
"pnl": 0.0
}
# Store for final CSV log
full_log_data = bet_data.copy()
full_log_data.update({
"marketId": market_id,
"marketName": meta.get("name"),
"timestamp": datetime.utcnow().isoformat()
})
self.bet_log.append(full_log_data)
# Store for active tracking
bets_to_track.append(bet_data)
processed_refs.add(str(bet_ref))
new_bets_found += 1
except Exception as e:
logger.warning(f"Failed to parse a matched bet fragment for {market_id}: {e}")
if new_bets_found > 0:
logger.info(f"[discovery] Found {new_bets_found} new matched bets for {market_id}")
except Exception as e:
logger.error(f"Failed to getMarketBets for {market_id}: {e}")
def _try_settle_discovered_bets(self, market_id: str, meta: Dict[str, Any], after_off: bool = False):
"""
Attempts to settle P/L for all discovered bets in a market.
"""
bets_to_track = meta.get("bets_to_track", [])
if not bets_to_track:
return
a_bet_was_settled_this_call = False
for b in bets_to_track:
if b.get("settled"):
continue
res = self._read_result_flag(b["selectionId"])
if res is not None:
# We have a result! Settle this bet.
b["settled"] = True
stake = b["stake"]
price = b["price"]
pnl = 0.0
if b["type"] == "LAY":
if str(res).strip().upper().startswith("WIN"):
pnl = - (price - 1.0) * stake # Loss
else: # Lose or Place
pnl = stake * (1.0 - COMMISSION_RATE) # Profit
elif b["type"] == "BACK":
if str(res).strip().upper().startswith("WIN"):
pnl = (price - 1.0) * stake * (1.0 - COMMISSION_RATE) # Profit
else: # Lose or Place
pnl = -stake # Loss
else:
logger.warning(f"Cannot calculate P&L for unknown bet type '{b['type']}'")
b["pnl"] = round(pnl, 2)
b["result"] = res
a_bet_was_settled_this_call = True
# Update global P&L trackers
self.running_pnl = round(self.running_pnl + pnl, 2)
self.virtual_bank = round(self.virtual_bank + pnl, 2)
logger.info(
f"[settled] mkt={market_id} sel={b['selectionId']} betRef={b['betRef']} "
f"type={b['type']} @{price:.2f} stake={stake:.2f} res={res} pnl={b['pnl']:.2f} | "
f"session_pnl={self.running_pnl:.2f} | bank={self.virtual_bank:.2f}"
)
# Write to the persistent P&L log
try:
self._update_pnl_csv(market_id, b["selectionId"], b['pnl'])
except Exception as e:
logger.warning(f"Failed to write P&L log: {e}")
elif after_off and not b.get("warned_unsettled"):
# No result yet, but we are past the close wait time
logger.warning(
f"[unsettled] mkt={market_id} sel={b['selectionId']} betRef={b['betRef']} "
f"is still unsettled after {POST_OFF_CLOSE_WAIT_MIN} mins. "
f"Waiting for 'result' stored value."
)
b["warned_unsettled"] = True # Only warn once
# --- Send Telegram Alert if market is fully settled ---
if not meta.get("telegram_alert_sent"):
all_settled = all(bet.get("settled") for bet in bets_to_track)
# Also check if we have actually discovered bets. Don't send alerts for empty markets.
bets_have_been_discovered = meta.get("last_bet_discovery_time") is not None
if all_settled and bets_have_been_discovered:
logger.info(f"Market {market_id} is fully settled. Sending Telegram alert.")
race_pnl = sum(b.get('pnl', 0.0) for b in bets_to_track)
race_name = meta.get('name', f'Market {market_id}')
start_dt = self.market_starts.get(market_id) or meta.get("start")
start_time_str = "Unknown Time"
if isinstance(start_dt, datetime):
try:
start_time_str = start_dt.astimezone().strftime('%H:%M')
except Exception:
start_time_str = start_dt.strftime('%H:%M UTC')
message = (
f"**Race Closed: {race_name} ({start_time_str})**\n"
f"--------------------\n"
f"**Race P&L:** £{race_pnl:.2f}\n"
f"**Session P&L:** £{self.running_pnl:.2f}"
)
self._send_telegram_alert(message)
meta["telegram_alert_sent"] = True
def run(self):
start_time = datetime.utcnow().replace(tzinfo=timezone.utc)
end_time = start_time + timedelta(hours=12) # Run for 12 hours
self.discover_markets()
last_discovery = time.time()
while datetime.utcnow().replace(tzinfo=timezone.utc) < end_time:
try:
if time.time() - last_discovery >= PERIODIC_DISCOVERY_SEC:
self.discover_markets()
last_discovery = time.time()
now = datetime.utcnow().replace(tzinfo=timezone.utc)
# Process open markets
for market_id, meta in list(self.open_markets.items()):
start = self.market_starts.get(market_id) or meta.get("start")
if not isinstance(start, datetime):
continue # Should have start time
secs_past_off = (now - start).total_seconds()
# 1. Check if we should close this market
after_off_check = secs_past_off / 60.0 >= POST_OFF_CLOSE_WAIT_MIN
if after_off_check:
# Try one last settle
self._try_settle_discovered_bets(market_id, meta, after_off=True)
# Check if all bets are settled before closing
all_settled = all(b.get("settled") for b in meta.get("bets_to_track", []))
if all_settled:
logger.info(f"Closing fully settled market {market_id}")
self.open_markets.pop(market_id, None)
continue # Move to next market
# 2. If market is active (post-off but not closed), scan for bets and results
if secs_past_off >= DISCOVER_BETS_AFTER_SEC:
# Scan for new matched bets periodically
last_scan = meta.get("last_bet_discovery_time")
if last_scan is None or (now - last_scan).total_seconds() >= REDISCOVER_BETS_FREQ_SEC:
self._discover_new_matched_bets(market_id, meta)
meta["last_bet_discovery_time"] = now
# Try to settle any outstanding bets
self._try_settle_discovered_bets(market_id, meta, after_off=after_off_check)
time.sleep(POLL_INTERVAL_SEC)
except KeyboardInterrupt:
logger.info("Interrupted...")
break
except Exception:
logger.error("Main loop error:\n" + traceback.format_exc())
time.sleep(3)
# --- Final settle and log dump ---
logger.info("Bot shutting down. Performing final settlement sweep...")
for mid, meta in list(self.open_markets.items()):
self._discover_new_matched_bets(mid, meta) # One last discovery
self._try_settle_discovered_bets(mid, meta, after_off=True)
try:
os.makedirs(BET_LOG_DIR, exist_ok=True)
out_csv = os.path.join(
BET_LOG_DIR, f"generic_bet_log_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.csv"
)
pd.DataFrame(self.bet_log).to_csv(out_csv, index=False)
logger.info(f"Full bet log saved: {out_csv} | final_session_pnl={self.running_pnl:.2f} | final_virtual_bank={self.virtual_bank:.2f}")
except Exception as e:
logger.warning(f"Failed to save bet log CSV: {e}")
try:
self._update_pnl_csv('SESSION_END', 'N/A', 0.0)
logger.info(f"Final P&L summary logged to {self.pnl_log_path}")
message = (
f"**Bot Shutting Down**\n"
f"--------------------\n"
f"**Final Session P&L:** £{self.running_pnl:.2f}\n"
f"**Final Virtual Bank:** £{self.virtual_bank:.2f}"
)
self._send_telegram_alert(message)
except Exception as e:
logger.warning(f"Failed to write final P&L log: {e}")
# ------------------------------ ENTRYPOINT --------------------------------
if __name__ == "__main__":
# API MUST BE CREATED FIRST
api = BetAngelAPI(BASE_URL)
# Apply Coupon and Rules at Startup
logger.info("Applying Betangel coupon 'Todays_Horses'...")
try:
api.apply_coupon()
except Exception as e:
logger.error(f"Failed to apply coupon at startup: {e}")
logger.info("Getting initial markets to apply 'Log Winner' rules file...")
try:
markets_json = api.get_markets()
markets_list = markets_json.get("result", {}).get("markets", [])
ids = [str(m.get("id")) for m in (markets_list or []) if m.get("id")]
if ids:
time.sleep(5) # Give BA a moment
api.apply_rules(ids)
logger.info(f"Applied rules to {len(ids)} markets.")
else:
logger.info("No markets found at startup to apply rules to.")
except Exception as e:
logger.error(f"Failed to get markets or apply rules at startup: {e}")
# Create and run the monitor
bot = BafPnLMonitor(api=api)
bot.run()