457 lines
16 KiB
Python
457 lines
16 KiB
Python
"""Sierra Chart TradeActivityLog parser.
|
|
|
|
Binary .data files are decoded field-by-field (TLV format):
|
|
[uint32 LE: field_id] [uint32 LE: length] [bytes * length]
|
|
|
|
Field 0x66 → SCDateTime (int64 LE, microseconds since 1899-12-30) ← timestamp
|
|
Field 0x67 → symbol string
|
|
Field 0x68 → log message text
|
|
|
|
parse_binary_log() returns [(datetime_or_None, str)] for each log entry.
|
|
All internal parsers work on this list; timestamps propagate to _make_trade.
|
|
|
|
Sim accounts (contain 'sim'/'simulated' in account name):
|
|
- Entry: User order entry ... AOE=true (not a Flatten)
|
|
- Entry price: first 'Trade simulation fill ... Last: X' after signal
|
|
- Direction: 'Modifying Attached Order from parent fill. Parent base price: B.
|
|
New price: P' → Long if P > B
|
|
- Exit type 1 (TP/SL): next simulation fill at price ≠ entry price
|
|
- Exit type 2 (Flatten): 'Flatten&CancelAllOrders | Last: X. Current Position
|
|
quantity: N' → exit = X, lots = |N|
|
|
|
|
Live accounts:
|
|
- Entry: Updated Internal Position Quantity to N. Previous: 0
|
|
- Entry price: first 'Modifying Attached Order from parent fill.
|
|
Parent base price: X' after entry update
|
|
- Exit: Updated Internal Position Quantity to 0. Previous: N
|
|
- Exit price (priority order):
|
|
1. 'Flatten&CancelAllOrders | Last: X'
|
|
2. 'Rithmic Direct - DTC (Filled)' + 'Limit' nearby → tp_price
|
|
3. 'Rithmic Direct - DTC (Filled)' + 'Stop' nearby → sl_price
|
|
4. 'User order entry | Last: X | AOE=true' (Reverse entry)
|
|
"""
|
|
|
|
import glob
|
|
import os
|
|
import re
|
|
import struct
|
|
from datetime import datetime, timedelta
|
|
|
|
_LOG_GLOB = "TradeActivityLog_*_UTC.{account}.data"
|
|
|
|
# SCDateTime: int64 microseconds since 1899-12-30
|
|
_SC_EPOCH = datetime(1899, 12, 30)
|
|
_SC_TS_MIN = int((datetime(2000, 1, 1) - _SC_EPOCH).total_seconds() * 1_000_000)
|
|
_SC_TS_MAX = int((datetime(2060, 1, 1) - _SC_EPOCH).total_seconds() * 1_000_000)
|
|
|
|
_FIELD_DATETIME = 0x66 # SCDateTime
|
|
_FIELD_MESSAGE = 0x68 # log message text
|
|
|
|
|
|
# ── File discovery ─────────────────────────────────────────────────────────────
|
|
|
|
def find_sc_log_dir():
|
|
"""Return first detected SC TradeActivityLogs directory, or None."""
|
|
patterns = [
|
|
"~/.local/share/bottles/bottles/*/drive_c/SierraChart*/TradeActivityLogs/",
|
|
"~/Games/SierraChart*/TradeActivityLogs/",
|
|
"~/Games/sierra-chart*/TradeActivityLogs/",
|
|
"~/.wine/drive_c/SierraChart*/TradeActivityLogs/",
|
|
"~/.PlayOnLinux/wineprefix/*/drive_c/SierraChart*/TradeActivityLogs/",
|
|
]
|
|
for pattern in patterns:
|
|
matches = glob.glob(os.path.expanduser(pattern))
|
|
if matches:
|
|
return matches[0]
|
|
return None
|
|
|
|
|
|
def find_log_files(log_dir, accounts):
|
|
"""Return [(filepath, account_name)] for all matching log files."""
|
|
if not log_dir or not os.path.isdir(log_dir):
|
|
return []
|
|
result = []
|
|
for account in accounts:
|
|
pattern = os.path.join(log_dir, _LOG_GLOB.format(account=account))
|
|
for path in glob.glob(pattern):
|
|
result.append((path, account))
|
|
return result
|
|
|
|
|
|
# ── Binary log parser ──────────────────────────────────────────────────────────
|
|
|
|
def parse_binary_log(filepath):
|
|
"""Read binary SC .data file.
|
|
|
|
Returns [(datetime_or_None, message_str)] — one entry per log line,
|
|
timestamped with the SCDateTime field that preceded it in the file.
|
|
"""
|
|
try:
|
|
with open(filepath, "rb") as f:
|
|
data = f.read()
|
|
except OSError:
|
|
return []
|
|
|
|
entries = []
|
|
i = 0
|
|
n = len(data)
|
|
cur_ts = None
|
|
|
|
while i + 8 <= n:
|
|
try:
|
|
field_id, field_len = struct.unpack_from("<II", data, i)
|
|
except struct.error:
|
|
break
|
|
i += 8
|
|
if i + field_len > n:
|
|
break
|
|
|
|
value = data[i : i + field_len]
|
|
i += field_len
|
|
|
|
if field_id == _FIELD_DATETIME and field_len == 8:
|
|
raw, = struct.unpack("<q", value)
|
|
cur_ts = (_SC_EPOCH + timedelta(microseconds=raw)
|
|
if _SC_TS_MIN < raw < _SC_TS_MAX else None)
|
|
|
|
elif field_id == _FIELD_MESSAGE and field_len > 0:
|
|
text = value.rstrip(b"\x00").decode("ascii", errors="replace")
|
|
if text.strip():
|
|
entries.append((cur_ts, text))
|
|
|
|
return entries
|
|
|
|
|
|
extract_strings = parse_binary_log # alias used by tray_app
|
|
|
|
|
|
# ── Public entry point ─────────────────────────────────────────────────────────
|
|
|
|
def parse_trades(entries, account, instruments, trade_date=None):
|
|
"""Parse all complete trades from [(datetime_or_None, str)] entry list."""
|
|
if _is_sim(account):
|
|
return _parse_sim(entries, account, instruments, trade_date)
|
|
return _parse_live(entries, account, instruments, trade_date)
|
|
|
|
|
|
# ── Shared helpers ─────────────────────────────────────────────────────────────
|
|
|
|
def _is_sim(account):
|
|
low = account.lower()
|
|
return "sim" in low or "simulated" in low
|
|
|
|
|
|
def _text(entry):
|
|
"""Extract the string from a (ts, str) entry."""
|
|
return entry[1]
|
|
|
|
|
|
def _ts(entry):
|
|
"""Extract the datetime (or None) from a (ts, str) entry."""
|
|
return entry[0]
|
|
|
|
|
|
def _extract_instrument(s, known):
|
|
m = re.search(r'\b([A-Z]{2,4})[A-Z]\d', s)
|
|
if m and m.group(1) in known:
|
|
return m.group(1)
|
|
return None
|
|
|
|
|
|
_PRICE = r"(\d+(?:\.\d+)?)" # matches 29744.25 but not a trailing bare period
|
|
|
|
|
|
def _ts_str(ts, trade_date=None):
|
|
"""Format a datetime for the API: actual time → HH:MM:SS; fallback → 00:00:00."""
|
|
if ts:
|
|
return ts.strftime("%Y-%m-%d %H:%M:%S")
|
|
if trade_date:
|
|
return f"{trade_date} 00:00:00"
|
|
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
|
|
def _find_direction(entries, start, n, window=40):
|
|
"""Long if first Modifying new_price > base, Short otherwise."""
|
|
for j in range(start, min(start + window, n)):
|
|
s = _text(entries[j])
|
|
if "Order modified internally" in s:
|
|
continue
|
|
m = re.search(
|
|
r"Modifying Attached Order from parent fill\."
|
|
r" Parent base price: " + _PRICE + r"\. New price: " + _PRICE, s)
|
|
if m:
|
|
return "Long" if float(m.group(2)) > float(m.group(1)) else "Short"
|
|
return None
|
|
|
|
|
|
def _nearby_order_type(entries, dtc_idx, n, window=4):
|
|
"""After DTC (Filled) at dtc_idx, look at nearby strings for order type."""
|
|
for k in range(dtc_idx + 1, min(dtc_idx + window, n)):
|
|
s = _text(entries[k])
|
|
if "Market" in s:
|
|
return "Market"
|
|
if "Limit" in s:
|
|
return "Limit"
|
|
if "Stop" in s:
|
|
return "Stop"
|
|
return None
|
|
|
|
|
|
def _make_trade(instrument, direction, lots, entry_price, exit_price,
|
|
account, instruments, entry_ts=None, exit_ts=None, trade_date=None):
|
|
cfg = instruments[instrument]
|
|
tick_size = cfg["tick_size"]
|
|
tick_value = cfg["tick_value"]
|
|
|
|
if direction == "Long":
|
|
pnl_ticks = round((exit_price - entry_price) / tick_size) * lots
|
|
else:
|
|
pnl_ticks = round((entry_price - exit_price) / tick_size) * lots
|
|
|
|
pnl = round(pnl_ticks * tick_value, 2)
|
|
|
|
entry_str = _ts_str(entry_ts, trade_date)
|
|
exit_str = _ts_str(exit_ts, trade_date)
|
|
|
|
if direction == "Long":
|
|
exec_str = (f"Buy {lots} @ {entry_price} / {entry_str},"
|
|
f"Sell {lots} @ {exit_price} / {exit_str}")
|
|
else:
|
|
exec_str = (f"Sell {lots} @ {entry_price} / {entry_str},"
|
|
f"Buy {lots} @ {exit_price} / {exit_str}")
|
|
|
|
return {
|
|
"trade_id": f"{account}_{instrument}_{entry_price}_{exit_price}",
|
|
"instrument": instrument,
|
|
"direction": direction,
|
|
"lots": lots,
|
|
"entry_price": entry_price,
|
|
"exit_price": exit_price,
|
|
"entry_datetime": entry_str,
|
|
"exit_datetime": exit_str,
|
|
"pnl": pnl,
|
|
"pnl_ticks": pnl_ticks,
|
|
"executions": exec_str,
|
|
"account": account,
|
|
}
|
|
|
|
|
|
# ── Sim parser ─────────────────────────────────────────────────────────────────
|
|
|
|
def _parse_sim(entries, account, instruments, trade_date=None):
|
|
trades = []
|
|
current_inst = None
|
|
i = 0
|
|
n = len(entries)
|
|
|
|
while i < n:
|
|
s = _text(entries[i])
|
|
|
|
inst = _extract_instrument(s, instruments)
|
|
if inst:
|
|
current_inst = inst
|
|
|
|
if (re.search(r"User order entry.+AOE=true", s)
|
|
and "Flatten" not in s
|
|
and current_inst):
|
|
|
|
entry_idx, entry_price = _sim_next_fill(entries, i + 1, n)
|
|
if entry_idx is None:
|
|
i += 1
|
|
continue
|
|
|
|
direction = _find_direction(entries, i + 1, n)
|
|
|
|
exit_idx, exit_price, lots, exit_direction = _sim_find_exit(
|
|
entries, entry_idx + 1, n, entry_price)
|
|
if exit_idx is None:
|
|
i += 1
|
|
continue
|
|
|
|
if direction is None:
|
|
direction = exit_direction
|
|
if direction is None:
|
|
i += 1
|
|
continue
|
|
|
|
if lots is None:
|
|
lots = 1
|
|
|
|
entry_ts = _ts(entries[entry_idx])
|
|
exit_ts = _ts(entries[exit_idx])
|
|
|
|
trades.append(_make_trade(
|
|
current_inst, direction, lots,
|
|
entry_price, exit_price, account, instruments,
|
|
entry_ts=entry_ts, exit_ts=exit_ts, trade_date=trade_date))
|
|
i = exit_idx + 1
|
|
continue
|
|
|
|
i += 1
|
|
return trades
|
|
|
|
|
|
def _sim_next_fill(entries, start, n, window=60):
|
|
"""Return (idx, last_price) of next simulation fill."""
|
|
for j in range(start, min(start + window, n)):
|
|
m = re.search(
|
|
r"simulation fill\. Bid: \S+ Ask: \S+ Last: " + _PRICE,
|
|
_text(entries[j]))
|
|
if m:
|
|
return j, float(m.group(1))
|
|
return None, None
|
|
|
|
|
|
def _sim_find_exit(entries, start, n, entry_price, window=400):
|
|
"""Find Sim trade exit. Returns (exit_idx, exit_price, lots, direction_or_None)."""
|
|
for j in range(start, min(start + window, n)):
|
|
s = _text(entries[j])
|
|
|
|
m_flat = re.search(
|
|
r"Flatten&CancelAllOrders \| Last: " + _PRICE
|
|
+ r".*?Current Position quantity: (-?\d+)", s)
|
|
if m_flat:
|
|
qty = int(m_flat.group(2))
|
|
direction = "Long" if qty > 0 else "Short"
|
|
return j, float(m_flat.group(1)), abs(qty), direction
|
|
|
|
m_fill = re.search(
|
|
r"simulation fill\. Bid: \S+ Ask: \S+ Last: " + _PRICE, s)
|
|
if m_fill:
|
|
price = float(m_fill.group(1))
|
|
if abs(price - entry_price) > 0.1:
|
|
fill_type = _nearby_order_type(entries, j, n)
|
|
if fill_type == "Limit":
|
|
direction = "Long" if price > entry_price else "Short"
|
|
elif fill_type == "Stop":
|
|
direction = "Long" if price < entry_price else "Short"
|
|
else:
|
|
direction = None
|
|
return j, price, None, direction
|
|
|
|
return None, None, None, None
|
|
|
|
|
|
# ── Live parser ────────────────────────────────────────────────────────────────
|
|
|
|
def _parse_live(entries, account, instruments, trade_date=None):
|
|
trades = []
|
|
current_inst = None
|
|
last_signal_price = None
|
|
i = 0
|
|
n = len(entries)
|
|
|
|
while i < n:
|
|
s = _text(entries[i])
|
|
|
|
inst = _extract_instrument(s, instruments)
|
|
if inst:
|
|
current_inst = inst
|
|
|
|
m_sig = re.search(
|
|
r"User order entry \| Last: " + _PRICE + r" \| AOE=true", s)
|
|
if m_sig and "Flatten" not in s:
|
|
last_signal_price = float(m_sig.group(1))
|
|
|
|
m_entry = re.search(
|
|
r"Updated Internal Position Quantity to (-?\d+)\. Previous: 0", s)
|
|
if m_entry and current_inst:
|
|
new_qty = int(m_entry.group(1))
|
|
if new_qty == 0:
|
|
i += 1
|
|
continue
|
|
|
|
direction = "Long" if new_qty > 0 else "Short"
|
|
lots = abs(new_qty)
|
|
entry_ts = _ts(entries[i])
|
|
|
|
entry_price, tp_price, sl_price = _live_bracket_prices(entries, i + 1, n)
|
|
|
|
if entry_price is None:
|
|
entry_price = last_signal_price
|
|
if entry_price is None:
|
|
i += 1
|
|
continue
|
|
|
|
exit_idx = _live_exit_idx(entries, i + 1, n)
|
|
if exit_idx is None:
|
|
i += 1
|
|
continue
|
|
|
|
exit_price = _live_exit_price(entries, i + 1, exit_idx, tp_price, sl_price, n)
|
|
if exit_price is None:
|
|
i += 1
|
|
continue
|
|
|
|
exit_ts = _ts(entries[exit_idx])
|
|
last_signal_price = None
|
|
|
|
trades.append(_make_trade(
|
|
current_inst, direction, lots,
|
|
entry_price, exit_price, account, instruments,
|
|
entry_ts=entry_ts, exit_ts=exit_ts, trade_date=trade_date))
|
|
i = exit_idx + 1
|
|
continue
|
|
|
|
i += 1
|
|
return trades
|
|
|
|
|
|
def _live_bracket_prices(entries, start, n, window=40):
|
|
"""Return (entry_price, tp_price, sl_price) from bracket modification strings."""
|
|
entry_price = None
|
|
bracket_new_prices = []
|
|
|
|
for j in range(start, min(start + window, n)):
|
|
s = _text(entries[j])
|
|
if "Order modified internally" in s:
|
|
continue
|
|
m = re.search(
|
|
r"Modifying Attached Order from parent fill\."
|
|
r" Parent base price: " + _PRICE + r"\. New price: " + _PRICE, s)
|
|
if m:
|
|
if entry_price is None:
|
|
entry_price = float(m.group(1))
|
|
bracket_new_prices.append(float(m.group(2)))
|
|
if len(bracket_new_prices) >= 2:
|
|
break
|
|
|
|
tp_price = bracket_new_prices[0] if len(bracket_new_prices) >= 1 else None
|
|
sl_price = bracket_new_prices[1] if len(bracket_new_prices) >= 2 else None
|
|
return entry_price, tp_price, sl_price
|
|
|
|
|
|
def _live_exit_idx(entries, start, n):
|
|
"""Find index of position returning to 0."""
|
|
for j in range(start, n):
|
|
if re.search(
|
|
r"Updated Internal Position Quantity to 0\. Previous: -?\d+",
|
|
_text(entries[j])):
|
|
return j
|
|
return None
|
|
|
|
|
|
def _live_exit_price(entries, start, exit_idx, tp_price, sl_price, n):
|
|
"""Determine exit price from events between entry and exit position updates."""
|
|
for j in range(start, exit_idx):
|
|
m = re.search(r"Flatten&CancelAllOrders \| Last: " + _PRICE,
|
|
_text(entries[j]))
|
|
if m:
|
|
return float(m.group(1))
|
|
|
|
for j in range(start, exit_idx):
|
|
if "Rithmic Direct - DTC (Filled)" in _text(entries[j]):
|
|
otype = _nearby_order_type(entries, j, exit_idx)
|
|
if otype == "Limit" and tp_price is not None:
|
|
return tp_price
|
|
if otype == "Stop" and sl_price is not None:
|
|
return sl_price
|
|
|
|
for j in range(start, exit_idx):
|
|
s = _text(entries[j])
|
|
m = re.search(r"User order entry \| Last: " + _PRICE + r" \| AOE=true", s)
|
|
if m and "Flatten" not in s:
|
|
return float(m.group(1))
|
|
|
|
return None
|