diff --git a/app/pt_security_logger.py b/app/pt_security_logger.py
new file mode 100644
index 00000000..6d0c9e0b
--- /dev/null
+++ b/app/pt_security_logger.py
@@ -0,0 +1,545 @@
+"""
+PowerTraderAI+ Security & Audit Logging
+Standalone security event logging module (does NOT extend pt_logging_system)
+providing:
+- Correlation ID context (thread-local, propagated through request chains)
+- Security event logging: API auth attempts, credential usage, suspicious activity
+- Dedicated audit log (separate file, rotation, secure permissions)
+- Structured JSON security events for SIEM integration
+
+Note: This module does NOT extend, import, or depend on pt_logging_system in
+any way. It is fully self-contained and manages its own logging handler
+independently, so security events are always written to a dedicated audit
+file regardless of the application's root logger configuration.
+
+Usage:
+    from pt_security_logger import get_security_logger, CorrelationContext
+
+    # Propagate correlation ID through a trading workflow
+    with CorrelationContext("trade-abc-123"):
+        get_security_logger().log_auth_attempt("robinhood", success=True)
+        get_security_logger().log_trade_event("BTC-USD", "buy", 0.1, 45000.0)
+
+    # Standalone
+    get_security_logger().log_suspicious_activity(
+        "rate_limit_exceeded", source_ip="10.0.0.1", details={"endpoint": "/orders"}
+    )
+"""
+
+import collections
+import json
+import logging
+import logging.handlers
+import os
+import stat
+import threading
+import time
+import uuid
+from dataclasses import dataclass, asdict
+from datetime import datetime, timezone
+from enum import Enum
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Security event types
+# ---------------------------------------------------------------------------
+class SecurityEventType(Enum):
+    AUTH_ATTEMPT = "auth_attempt"  # API key authentication attempt
+    AUTH_SUCCESS = "auth_success"  # Successful authentication
+    AUTH_FAILURE = "auth_failure"  # Failed authentication
+    CREDENTIAL_USE = "credential_use"  # Credential accessed/used
+    CREDENTIAL_ROTATION = "credential_rotation"  # Credential rotated
+    SUSPICIOUS_ACTIVITY = "suspicious_activity"  # Anomalous behavior detected
+    PERMISSION_DENIED = "permission_denied"  # Insufficient API permissions
+    RATE_LIMIT = "rate_limit"  # Rate limit hit
+    TRADE_EXECUTED = "trade_executed"  # Order placed
+    TRADE_REJECTED = "trade_rejected"  # Order rejected
+    CONFIG_CHANGE = "config_change"  # Configuration modified
+    SYSTEM_START = "system_start"  # Application started
+    SYSTEM_STOP = "system_stop"  # Application stopped
+
+
+@dataclass
+class SecurityEvent:
+    """Structured security event for audit trail."""
+
+    event_id: str
+    event_type: str
+    timestamp: float
+    message: str
+    correlation_id: Optional[str]
+    source: Optional[str] = None
+    user_id: Optional[str] = None
+    source_ip: Optional[str] = None
+    success: Optional[bool] = None
+    details: Optional[Dict[str, Any]] = None
+
+    def to_dict(self) -> dict:
+        return asdict(self)
+
+    def to_json(self) -> str:
+        return json.dumps(self.to_dict(), default=str)
+
+    @staticmethod
+    def new_id() -> str:
+        # UTC timestamp in ID so audit records are timezone-unambiguous
+        ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
+        return f"SEC_{ts}_{uuid.uuid4().hex[:8]}"
+
+
+# ---------------------------------------------------------------------------
+# Correlation ID context (thread-local)
+# ---------------------------------------------------------------------------
+_correlation_local = threading.local()
+
+
+def get_correlation_id() -> Optional[str]:
+    """Return the current thread's correlation ID, or None if not set."""
+    return getattr(_correlation_local, "correlation_id", None)
+
+
+def set_correlation_id(cid: str) -> None:
+    """Set the correlation ID for the current thread."""
+    _correlation_local.correlation_id = cid
+
+
+def clear_correlation_id() -> None:
+    """Clear the correlation ID for the current thread."""
+    try:
+        del _correlation_local.correlation_id
+    except AttributeError:
+        pass
+
+
+class CorrelationContext:
+    """
+    Context manager that sets a correlation ID for the current thread,
+    then restores the previous value on exit.
+
+    Each call to __enter__ saves the previous ID on a stack, so nested
+    and reused instances are safe.
+
+    Usage:
+        with CorrelationContext("trade-workflow-xyz"):
+            process_order()  # all logs within will carry this correlation ID
+    """
+
+    def __init__(self, correlation_id: Optional[str] = None):
+        self._cid = correlation_id or f"CID_{uuid.uuid4().hex[:12]}"
+        self._stack: List[Optional[str]] = []
+
+    def __enter__(self) -> str:
+        self._stack.append(get_correlation_id())
+        set_correlation_id(self._cid)
+        return self._cid
+
+    def __exit__(self, *_) -> None:
+        if self._stack:
+            previous = self._stack.pop()
+            if previous is not None:
+                set_correlation_id(previous)
+            else:
+                clear_correlation_id()
+
+
+# Keep snake_case alias for backwards compatibility
+correlation_context = CorrelationContext
+
+
+# ---------------------------------------------------------------------------
+# CorrelationLogFilter — injects correlation_id into log records
+# ---------------------------------------------------------------------------
+class CorrelationLogFilter(logging.Filter):
+    """
+    Logging filter that injects the current thread's correlation ID
+    into every LogRecord. Works with any handler/formatter.
+    """
+
+    def filter(self, record: logging.LogRecord) -> bool:
+        record.correlation_id = get_correlation_id() or ""
+        return True
+
+
+# ---------------------------------------------------------------------------
+# SecureRotatingFileHandler — chmod backup files after rotation
+# ---------------------------------------------------------------------------
+class _SecureRotatingFileHandler(logging.handlers.RotatingFileHandler):
+    """RotatingFileHandler subclass that applies owner-only permissions after rollover."""
+
+    def doRollover(self) -> None:
+        super().doRollover()
+        # Re-secure the new active file and all existing backups
+        for path in self._audit_paths():
+            _chmod_secure(path)
+
+    def _audit_paths(self):
+        yield self.baseFilename
+        for n in range(1, self.backupCount + 1):
+            candidate = f"{self.baseFilename}.{n}"
+            if os.path.exists(candidate):
+                yield candidate
+
+
+def _chmod_secure(path: str) -> None:
+    """Set owner-only read/write on path; log warning on failure."""
+    try:
+        if os.path.exists(path):
+            os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
+    except OSError as exc:
+        logger.warning(
+            "Could not secure audit log permissions on %s: %s — "
+            "file may have permissive permissions",
+            path,
+            exc,
+        )
+
+
+# ---------------------------------------------------------------------------
+# SecurityLogger
+# ---------------------------------------------------------------------------
+class SecurityLogger:
+    """
+    Application-wide security and audit logger.
+
+    Writes structured JSON security events to a dedicated audit log file
+    (separate from the main application log). The audit log uses a
+    RotatingFileHandler with secure (owner-only) file permissions.
+
+    Args:
+        log_dir: Directory for the audit log file.  Defaults to
+            ~/.powertraderai/logs so the source tree is never polluted.
+    """
+
+    AUDIT_LOG_FILENAME = "security_audit.jsonl"
+    MAX_BYTES = 10 * 1024 * 1024  # 10 MB per file
+    BACKUP_COUNT = 10  # Keep 10 rotated files
+
+    def __init__(self, log_dir: Optional[str] = None):
+        self._log_dir = log_dir or os.path.join(
+            os.path.expanduser("~"), ".powertraderai", "logs"
+        )
+        os.makedirs(self._log_dir, exist_ok=True)
+        self._audit_path = os.path.join(self._log_dir, self.AUDIT_LOG_FILENAME)
+        self._lock = threading.Lock()
+        # Use a private logger instance (not the global registry) to avoid the
+        # shared-handler pitfall when two SecurityLogger instances use the same dir.
+        self._audit_logger = logging.Logger(
+            f"pt.security.audit.{id(self)}", level=logging.DEBUG
+        )
+        self._audit_logger.propagate = False
+        self._handler = self._create_handler()
+        self._audit_logger.addHandler(self._handler)
+
+    def _create_handler(self) -> _SecureRotatingFileHandler:
+        handler = _SecureRotatingFileHandler(
+            self._audit_path,
+            maxBytes=self.MAX_BYTES,
+            backupCount=self.BACKUP_COUNT,
+            encoding="utf-8",
+        )
+        handler.setFormatter(logging.Formatter("%(message)s"))  # Raw JSON lines
+        handler.addFilter(CorrelationLogFilter())
+        # Secure permissions once at creation
+        _chmod_secure(self._audit_path)
+        return handler
+
+    def close(self) -> None:
+        """Flush and close the underlying handler (call on shutdown).
+        Idempotent: safe to call multiple times."""
+        if self._handler is None:
+            return
+        handler = self._handler
+        self._handler = None
+        try:
+            handler.flush()
+            handler.close()
+        finally:
+            self._audit_logger.removeHandler(handler)
+
+    def _emit(self, event: SecurityEvent) -> None:
+        """Write security event to audit log (thread-safe)."""
+        with self._lock:
+            try:
+                self._audit_logger.info(event.to_json())
+            except Exception as exc:
+                logger.error("Failed to write security audit event: %s", exc)
+
+    def _make_event(
+        self,
+        event_type: SecurityEventType,
+        message: str,
+        source: Optional[str] = None,
+        success: Optional[bool] = None,
+        details: Optional[Dict[str, Any]] = None,
+        source_ip: Optional[str] = None,
+        user_id: Optional[str] = None,
+    ) -> SecurityEvent:
+        return SecurityEvent(
+            event_id=SecurityEvent.new_id(),
+            event_type=event_type.value,
+            timestamp=time.time(),
+            message=message,
+            correlation_id=get_correlation_id(),
+            source=source,
+            success=success,
+            details=details,
+            source_ip=source_ip,
+            user_id=user_id,
+        )
+
+    # ------------------------------------------------------------------
+    # Public logging methods
+    # ------------------------------------------------------------------
+    def log_auth_attempt(
+        self,
+        api_name: str,
+        success: bool,
+        user_id: Optional[str] = None,
+        source_ip: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log an API authentication attempt (success or failure)."""
+        event_type = (
+            SecurityEventType.AUTH_SUCCESS
+            if success
+            else SecurityEventType.AUTH_FAILURE
+        )
+        msg = f"API auth {'succeeded' if success else 'FAILED'} for {api_name}"
+        if not success:
+            logger.warning("SECURITY: %s", msg)
+        self._emit(
+            self._make_event(
+                event_type,
+                msg,
+                source=api_name,
+                success=success,
+                details=details,
+                source_ip=source_ip,
+                user_id=user_id,
+            )
+        )
+
+    def log_credential_use(
+        self,
+        api_name: str,
+        operation: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log that credentials were accessed/used for an operation."""
+        msg = f"Credential used: {api_name} for {operation}"
+        self._emit(
+            self._make_event(
+                SecurityEventType.CREDENTIAL_USE,
+                msg,
+                source=api_name,
+                success=True,
+                details={**(details or {}), "operation": operation},
+            )
+        )
+
+    def log_credential_rotation(
+        self,
+        api_name: str,
+        success: bool,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log a credential rotation event."""
+        msg = (
+            f"Credential rotation {'succeeded' if success else 'FAILED'} for {api_name}"
+        )
+        if not success:
+            logger.critical("SECURITY: %s", msg)
+        self._emit(
+            self._make_event(
+                SecurityEventType.CREDENTIAL_ROTATION,
+                msg,
+                source=api_name,
+                success=success,
+                details=details,
+            )
+        )
+
+    def log_suspicious_activity(
+        self,
+        activity_type: str,
+        source_ip: Optional[str] = None,
+        user_id: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log detected suspicious activity."""
+        msg = f"Suspicious activity detected: {activity_type}"
+        logger.warning("SECURITY ALERT: %s", msg)
+        self._emit(
+            self._make_event(
+                SecurityEventType.SUSPICIOUS_ACTIVITY,
+                msg,
+                source=activity_type,
+                success=False,
+                source_ip=source_ip,
+                user_id=user_id,
+                details=details,
+            )
+        )
+
+    def log_permission_denied(
+        self,
+        api_name: str,
+        required_permission: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log an API permission denial."""
+        msg = f"Permission denied: {required_permission} on {api_name}"
+        logger.error("SECURITY: %s", msg)
+        self._emit(
+            self._make_event(
+                SecurityEventType.PERMISSION_DENIED,
+                msg,
+                source=api_name,
+                success=False,
+                details={**(details or {}), "required_permission": required_permission},
+            )
+        )
+
+    def log_trade_event(
+        self,
+        symbol: str,
+        side: str,
+        quantity: float,
+        price: float,
+        success: bool = True,
+        order_id: Optional[str] = None,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log a trade execution event to the audit trail."""
+        event_type = (
+            SecurityEventType.TRADE_EXECUTED
+            if success
+            else SecurityEventType.TRADE_REJECTED
+        )
+        msg = f"Trade {'executed' if success else 'REJECTED'}: {side} {quantity} {symbol} @ {price}"
+        self._emit(
+            self._make_event(
+                event_type,
+                msg,
+                success=success,
+                details={
+                    "symbol": symbol,
+                    "side": side,
+                    "quantity": quantity,
+                    "price": price,
+                    "order_id": order_id,
+                    **(details or {}),
+                },
+            )
+        )
+
+    def log_rate_limit(
+        self,
+        api_name: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log a rate limit event."""
+        msg = f"Rate limit hit on {api_name}"
+        logger.warning("SECURITY: %s", msg)
+        self._emit(
+            self._make_event(
+                SecurityEventType.RATE_LIMIT,
+                msg,
+                source=api_name,
+                success=False,
+                details=details,
+            )
+        )
+
+    def log_config_change(
+        self,
+        component: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log a configuration change event."""
+        msg = f"Configuration changed: {component}"
+        logger.info("SECURITY: %s", msg)
+        self._emit(
+            self._make_event(
+                SecurityEventType.CONFIG_CHANGE,
+                msg,
+                source=component,
+                details=details,
+            )
+        )
+
+    def log_system_event(
+        self,
+        event_type: SecurityEventType,
+        message: str,
+        details: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Log a system lifecycle event (start, stop, config change, rate limit)."""
+        self._emit(self._make_event(event_type, message, details=details))
+
+    def get_recent_events(self, limit: int = 100) -> List[dict]:
+        """
+        Read the last `limit` events from the audit log.
+
+        Time complexity is O(file_size) — the iterator walks every line;
+        memory is bounded to O(limit) via ``collections.deque(maxlen=limit)``.
+        Only reads the active log file; rotated backups (`*.1`, `*.2`, …)
+        are not included.
+
+        Returns [] when limit <= 0.
+        """
+        if limit <= 0:
+            return []
+        if not os.path.exists(self._audit_path):
+            return []
+        try:
+            with open(self._audit_path, "r", encoding="utf-8") as f:
+                tail = collections.deque(f, maxlen=limit)
+            events = []
+            for line in tail:
+                line = line.strip()
+                if line:
+                    try:
+                        events.append(json.loads(line))
+                    except json.JSONDecodeError:
+                        pass
+            return events
+        except OSError:
+            return []
+
+
+# ---------------------------------------------------------------------------
+# Module-level lazy singleton
+# ---------------------------------------------------------------------------
+_security_logger_instance: Optional[SecurityLogger] = None
+_security_logger_lock = threading.Lock()
+
+
+def get_security_logger(log_dir: Optional[str] = None) -> SecurityLogger:
+    """
+    Return the application-wide SecurityLogger singleton.
+
+    Lazy: the logger (and its audit file) is only created on first call,
+    not at module import time.  Pass log_dir only on the first call;
+    subsequent calls return the existing instance.
+    """
+    global _security_logger_instance
+    if _security_logger_instance is None:
+        with _security_logger_lock:
+            if _security_logger_instance is None:
+                _security_logger_instance = SecurityLogger(log_dir=log_dir)
+    return _security_logger_instance
+
+
+# Convenience alias — module-level name that lazily resolves on first attribute access
+class _LazySecurityLogger:
+    """Proxy that forwards all attribute access to get_security_logger()."""
+
+    def __getattr__(self, name: str):
+        return getattr(get_security_logger(), name)
+
+
+security_logger = _LazySecurityLogger()
diff --git a/app/test_security_logger.py b/app/test_security_logger.py
new file mode 100644
index 00000000..c180edd6
--- /dev/null
+++ b/app/test_security_logger.py
@@ -0,0 +1,297 @@
+"""Tests for pt_security_logger - issue #55."""
+
+import json
+import os
+import shutil
+import tempfile
+import threading
+import time
+import unittest
+
+from pt_security_logger import (
+    CorrelationContext,
+    SecurityEvent,
+    SecurityEventType,
+    SecurityLogger,
+    clear_correlation_id,
+    correlation_context,
+    get_correlation_id,
+    set_correlation_id,
+)
+
+
+class TestCorrelationID(unittest.TestCase):
+    def tearDown(self):
+        clear_correlation_id()
+
+    def test_default_is_none(self):
+        clear_correlation_id()
+        self.assertIsNone(get_correlation_id())
+
+    def test_set_and_get(self):
+        set_correlation_id("test-cid-123")
+        self.assertEqual(get_correlation_id(), "test-cid-123")
+
+    def test_context_manager_sets_id(self):
+        with CorrelationContext("ctx-abc") as cid:
+            self.assertEqual(cid, "ctx-abc")
+            self.assertEqual(get_correlation_id(), "ctx-abc")
+
+    def test_context_manager_restores_previous(self):
+        set_correlation_id("outer")
+        with CorrelationContext("inner"):
+            self.assertEqual(get_correlation_id(), "inner")
+        self.assertEqual(get_correlation_id(), "outer")
+
+    def test_context_manager_clears_when_no_previous(self):
+        clear_correlation_id()
+        with CorrelationContext("temp"):
+            pass
+        self.assertIsNone(get_correlation_id())
+
+    def test_context_manager_auto_generates_id(self):
+        with CorrelationContext() as cid:
+            self.assertIsNotNone(cid)
+            self.assertTrue(cid.startswith("CID_"))
+
+    def test_context_manager_nested_safe(self):
+        """Reusing the same CorrelationContext instance in nested with-blocks is safe."""
+        ctx = CorrelationContext("reused")
+        with ctx as cid1:
+            self.assertEqual(cid1, "reused")
+            with ctx as cid2:
+                self.assertEqual(cid2, "reused")
+            # Inner exit must restore the outer's saved state
+            self.assertEqual(get_correlation_id(), "reused")
+        self.assertIsNone(get_correlation_id())
+
+    def test_snake_case_alias_works(self):
+        """correlation_context alias must still function for backwards compatibility."""
+        with correlation_context("alias-test") as cid:
+            self.assertEqual(cid, "alias-test")
+
+    def test_thread_local_isolation(self):
+        """Each thread has its own correlation ID."""
+        results = {}
+
+        def worker(name, cid):
+            set_correlation_id(cid)
+            time.sleep(0.01)
+            results[name] = get_correlation_id()
+
+        t1 = threading.Thread(target=worker, args=("t1", "cid-thread-1"))
+        t2 = threading.Thread(target=worker, args=("t2", "cid-thread-2"))
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+        self.assertEqual(results["t1"], "cid-thread-1")
+        self.assertEqual(results["t2"], "cid-thread-2")
+
+
+class TestSecurityEvent(unittest.TestCase):
+    def test_new_id_format(self):
+        eid = SecurityEvent.new_id()
+        self.assertTrue(eid.startswith("SEC_"))
+        self.assertGreater(len(eid), 10)
+
+    def test_new_id_uses_utc(self):
+        """Event ID must be UTC-based (no timezone-ambiguous local time)."""
+        from datetime import datetime, timezone
+
+        before = datetime.now(timezone.utc).strftime("%Y%m%d")
+        eid = SecurityEvent.new_id()
+        self.assertIn(before, eid)
+
+    def test_to_dict_has_required_fields(self):
+        event = SecurityEvent(
+            event_id="SEC_001",
+            event_type=SecurityEventType.AUTH_ATTEMPT.value,
+            timestamp=time.time(),
+            message="test",
+            correlation_id=None,
+        )
+        d = event.to_dict()
+        for field in ("event_id", "event_type", "timestamp", "message"):
+            self.assertIn(field, d)
+
+    def test_to_json_valid(self):
+        event = SecurityEvent(
+            event_id="SEC_002",
+            event_type=SecurityEventType.AUTH_SUCCESS.value,
+            timestamp=time.time(),
+            message="auth ok",
+            correlation_id="cid-test",
+        )
+        parsed = json.loads(event.to_json())
+        self.assertEqual(parsed["correlation_id"], "cid-test")
+
+
+class TestSecurityLogger(unittest.TestCase):
+    def setUp(self):
+        self.tmpdir = tempfile.mkdtemp()
+        self.sec_logger = SecurityLogger(log_dir=self.tmpdir)
+        clear_correlation_id()
+
+    def tearDown(self):
+        clear_correlation_id()
+        # close() flushes and releases file handles (critical on Windows)
+        self.sec_logger.close()
+        shutil.rmtree(self.tmpdir, ignore_errors=True)
+
+    def _events(self):
+        return self.sec_logger.get_recent_events()
+
+    def test_log_auth_attempt_success(self):
+        self.sec_logger.log_auth_attempt("robinhood", success=True)
+        events = self._events()
+        self.assertEqual(len(events), 1)
+        self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_SUCCESS.value)
+        self.assertTrue(events[0]["success"])
+
+    def test_log_auth_attempt_failure(self):
+        self.sec_logger.log_auth_attempt("robinhood", success=False)
+        events = self._events()
+        self.assertEqual(events[0]["event_type"], SecurityEventType.AUTH_FAILURE.value)
+        self.assertFalse(events[0]["success"])
+
+    def test_log_credential_use(self):
+        self.sec_logger.log_credential_use("robinhood", "place_order")
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.CREDENTIAL_USE.value
+        )
+        self.assertEqual(events[0]["details"]["operation"], "place_order")
+
+    def test_log_credential_rotation(self):
+        self.sec_logger.log_credential_rotation("robinhood", success=True)
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.CREDENTIAL_ROTATION.value
+        )
+
+    def test_log_suspicious_activity(self):
+        self.sec_logger.log_suspicious_activity(
+            "rate_limit_exceeded",
+            source_ip="1.2.3.4",
+            details={"endpoint": "/orders", "count": 100},
+        )
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.SUSPICIOUS_ACTIVITY.value
+        )
+        self.assertEqual(events[0]["source_ip"], "1.2.3.4")
+
+    def test_log_permission_denied(self):
+        self.sec_logger.log_permission_denied("robinhood", "sell")
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.PERMISSION_DENIED.value
+        )
+        self.assertEqual(events[0]["details"]["required_permission"], "sell")
+
+    def test_log_trade_event_success(self):
+        self.sec_logger.log_trade_event(
+            "BTC-USD", "buy", 0.1, 45000.0, order_id="ord-001"
+        )
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.TRADE_EXECUTED.value
+        )
+        self.assertEqual(events[0]["details"]["symbol"], "BTC-USD")
+
+    def test_log_trade_event_rejected(self):
+        self.sec_logger.log_trade_event("ETH-USD", "sell", 1.0, 3000.0, success=False)
+        events = self._events()
+        self.assertEqual(
+            events[0]["event_type"], SecurityEventType.TRADE_REJECTED.value
+        )
+
+    def test_log_rate_limit(self):
+        self.sec_logger.log_rate_limit("binance", details={"endpoint": "/api/v3/order"})
+        events = self._events()
+        self.assertEqual(events[0]["event_type"], SecurityEventType.RATE_LIMIT.value)
+        self.assertEqual(events[0]["source"], "binance")
+
+    def test_log_config_change(self):
+        self.sec_logger.log_config_change("risk_limits", details={"max_position": 0.1})
+        events = self._events()
+        self.assertEqual(events[0]["event_type"], SecurityEventType.CONFIG_CHANGE.value)
+        self.assertEqual(events[0]["source"], "risk_limits")
+
+    def test_correlation_id_captured_in_event(self):
+        with CorrelationContext("trade-workflow-xyz"):
+            self.sec_logger.log_auth_attempt("robinhood", success=True)
+        events = self._events()
+        self.assertEqual(events[0]["correlation_id"], "trade-workflow-xyz")
+
+    def test_correlation_id_none_outside_context(self):
+        clear_correlation_id()
+        self.sec_logger.log_auth_attempt("robinhood", success=True)
+        events = self._events()
+        self.assertIsNone(events[0]["correlation_id"])
+
+    def test_multiple_events_ordered(self):
+        self.sec_logger.log_auth_attempt("robinhood", success=True)
+        self.sec_logger.log_credential_use("robinhood", "fetch_positions")
+        self.sec_logger.log_trade_event("BTC-USD", "buy", 0.5, 44000.0)
+        events = self._events()
+        self.assertEqual(len(events), 3)
+        # Assert event_type append order — robust against clock adjustments
+        # (time.time() can move backwards on NTP correction or DST shifts).
+        self.assertEqual(
+            [e["event_type"] for e in events],
+            [
+                SecurityEventType.AUTH_SUCCESS.value,
+                SecurityEventType.CREDENTIAL_USE.value,
+                SecurityEventType.TRADE_EXECUTED.value,
+            ],
+        )
+        # Sanity check: timestamps are non-decreasing (cheap monotonicity
+        # check; sub-second jumps backwards from NTP do not break this).
+        timestamps = [e["timestamp"] for e in events]
+        for prev, curr in zip(timestamps, timestamps[1:]):
+            self.assertLessEqual(prev, curr)
+
+    def test_audit_file_created(self):
+        self.sec_logger.log_auth_attempt("robinhood", success=True)
+        self.assertTrue(os.path.exists(self.sec_logger._audit_path))
+
+    def test_get_recent_events_empty_when_no_log(self):
+        """get_recent_events() returns [] when no audit log file exists."""
+        tmpdir2 = tempfile.mkdtemp()
+        try:
+            sl = SecurityLogger(log_dir=tmpdir2)
+            # Close immediately — no events written, so file may not exist yet
+            sl.close()
+            # Remove the file if it was created
+            audit_path = sl._audit_path
+            if os.path.exists(audit_path):
+                os.remove(audit_path)
+            # Now verify empty result
+            self.assertEqual(sl.get_recent_events(), [])
+        finally:
+            shutil.rmtree(tmpdir2, ignore_errors=True)
+
+    def test_get_recent_events_limit(self):
+        """get_recent_events(limit=N) returns exactly the last N events."""
+        for i in range(10):
+            self.sec_logger.log_auth_attempt("api", success=True)
+        events = self.sec_logger.get_recent_events(limit=5)
+        self.assertEqual(len(events), 5)
+        # Should be the last 5 — timestamps are non-decreasing
+        all_events = self.sec_logger.get_recent_events(limit=100)
+        self.assertEqual(events, all_events[-5:])
+
+    def test_log_system_event(self):
+        self.sec_logger.log_system_event(
+            SecurityEventType.SYSTEM_START,
+            "PowerTraderAI started",
+            details={"version": "1.0"},
+        )
+        events = self._events()
+        self.assertEqual(events[0]["event_type"], SecurityEventType.SYSTEM_START.value)
+
+
+if __name__ == "__main__":
+    unittest.main()