Add trade ledger debugging exercise

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
SirBlobandClaude Opus 5.5 committed 2026-10-31 09:03:25 -04:00
commit 0cd0fdb641
6 files changed
+358

No files matched your search

+68
View File
@@ -0,0 +1,68 @@
import sys
import time
LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR")
class Record:
__slots__ = ("ts", "level", "msg", "context")
def __init__(self, ts, level, msg, context):
self.ts = ts
self.level = level
self.msg = msg
self.context = context
def render(self):
ctx = " ".join(f"{k}={v}" for k, v in sorted(self.context.items()))
return f"{self.ts:.6f} {self.level:<7} {self.msg} {ctx}".rstrip()
class BufferedLogger:
def __init__(self, name, level="INFO", stream=None, capacity=64):
self.name = name
self.stream = stream if stream is not None else sys.stderr
self.capacity = capacity
self._context = {}
self._buffer = []
self.set_level(level)
def set_level(self, level):
level = level.upper()
if level not in LEVELS:
raise ValueError(f"unknown level {level!r}")
self.level = level
def push(self, **fields):
self._context.update(fields)
def clear(self):
self._context.clear()
def _enabled(self, level):
return level >= self.level
def _log(self, level, msg):
if not self._enabled(level):
return
self._buffer.append(Record(time.time(), level, msg, self._context))
if len(self._buffer) >= self.capacity:
self.flush()
def debug(self, msg):
self._log("DEBUG", msg)
def info(self, msg):
self._log("INFO", msg)
def warning(self, msg):
self._log("WARNING", msg)
def error(self, msg):
self._log("ERROR", msg)
def flush(self):
for rec in self._buffer:
self.stream.write(f"[{self.name}] {rec.render()}\n")
self.stream.flush()
self._buffer.clear()