!pip install schedule apscheduler --quiet

import time
import logging
import threading
import random
from datetime import datetime, timedelta
from typing import Callable, Dict, List, Optional, Any
from dataclasses import dataclass, field

import schedule
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger

# Configure logging so every scheduled task prints a timestamped line
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s  %(levelname)-8s  %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S",
)
logger = logging.getLogger("trading_scheduler")

print("Imports loaded.")

def fetch_market_data(symbols: List[str]) -> Dict[str, float]:
    """
    Fetch the latest price for each symbol in the watchlist.

    In production, replace the random price generation with a real data
    provider call, for example:
        import yfinance as yf
        return {s: yf.Ticker(s).fast_info['last_price'] for s in symbols}

    Parameters
    ----------
    symbols : list of str
        Ticker symbols to fetch, e.g. ['AAPL', 'MSFT', 'SPY'].

    Returns
    -------
    dict
        Mapping of symbol -> latest price (float).

    Example
    -------
    >>> prices = fetch_market_data(['AAPL', 'MSFT'])
    >>> print(prices)
    {'AAPL': 182.34, 'MSFT': 415.67}
    """

    # Simulate realistic price ranges per symbol
    base_prices = {
        "AAPL" : 185.0,
        "MSFT" : 420.0,
        "GOOGL": 175.0,
        "SPY"  : 530.0,
        "BTC"  : 67000.0,
    }

    prices = {}
    for symbol in symbols:
        base  = base_prices.get(symbol, 100.0)
        # Add small random noise to simulate live price movement
        noise = random.uniform(-base * 0.005, base * 0.005)
        prices[symbol] = round(base + noise, 2)

    logger.info("fetch_market_data | fetched %d symbols: %s", len(prices), prices)
    return prices


# Quick test
fetch_market_data(["AAPL", "MSFT", "SPY"])

def generate_signals(prices: Dict[str, float]) -> Dict[str, str]:
    """
    Generate simple BUY / HOLD / SELL signals from current prices.

    Uses a basic momentum heuristic: compare the last two digits of the
    price to a threshold. In production, replace this with a real strategy
    such as moving average crossovers, RSI thresholds, or an ML model.

    Parameters
    ----------
    prices : dict
        Symbol -> price mapping from fetch_market_data().

    Returns
    -------
    dict
        Symbol -> signal string, one of 'BUY', 'SELL', or 'HOLD'.

    Example
    -------
    >>> signals = generate_signals({'AAPL': 182.34, 'MSFT': 415.67})
    >>> print(signals)
    {'AAPL': 'HOLD', 'MSFT': 'BUY'}
    """

    signals = {}
    for symbol, price in prices.items():
        # Use the fractional cents as a pseudo-random seed for demo signals
        remainder = int(price * 100) % 10

        if remainder <= 2:
            signals[symbol] = "BUY"
        elif remainder >= 8:
            signals[symbol] = "SELL"
        else:
            signals[symbol] = "HOLD"

    logger.info("generate_signals | signals: %s", signals)
    return signals


# Quick test
prices  = fetch_market_data(["AAPL", "MSFT", "SPY"])
signals = generate_signals(prices)
print(signals)

def execute_orders(signals: Dict[str, str], dry_run: bool = True) -> List[Dict[str, Any]]:
    """
    Submit orders for any symbols with an actionable BUY or SELL signal.

    When dry_run=True (the default), orders are logged but never sent to a
    broker.  Set dry_run=False only when connected to a paper or live
    trading account.

    Parameters
    ----------
    signals : dict
        Symbol -> signal string from generate_signals().
    dry_run : bool, optional
        If True, simulate order submission without calling any broker API.
        Default: True.

    Returns
    -------
    list of dict
        Each dict represents a submitted (or simulated) order:
        {'symbol', 'side', 'qty', 'status', 'timestamp'}.

    Example
    -------
    >>> orders = execute_orders({'AAPL': 'BUY', 'MSFT': 'HOLD'})
    >>> print(orders)
    [{'symbol': 'AAPL', 'side': 'BUY', 'qty': 1, 'status': 'simulated', ...}]
    """

    submitted = []

    for symbol, signal in signals.items():
        # HOLD signals require no action
        if signal == "HOLD":
            continue

        order = {
            "symbol"   : symbol,
            "side"     : signal,
            "qty"      : 1,  # Fixed quantity for demonstration
            "status"   : "simulated" if dry_run else "submitted",
            "timestamp": datetime.now().isoformat(),
        }

        submitted.append(order)

        if dry_run:
            logger.info("execute_orders | DRY RUN %s %s qty=1", signal, symbol)
        else:
            # Replace this block with your broker's order API, e.g.:
            # alpaca.submit_order(symbol=symbol, qty=1, side=signal.lower(), ...)
            logger.info("execute_orders | SUBMITTED %s %s qty=1", signal, symbol)

    return submitted


# Quick test
orders = execute_orders({"AAPL": "BUY", "MSFT": "HOLD", "SPY": "SELL"})
print(orders)

def generate_daily_report(symbols: List[str]) -> Dict[str, Any]:
    """
    Produce an end-of-day summary report for the watchlist.

    Fetches current prices, computes simulated daily P&L, and returns a
    summary dict.  In production, compare closing price against an opening
    price stored in a database or fetched from a historical data provider.

    Parameters
    ----------
    symbols : list of str
        Ticker symbols to include in the report.

    Returns
    -------
    dict
        {
          'date'        : str  — report date (YYYY-MM-DD),
          'symbols'     : list — tickers covered,
          'prices'      : dict — closing prices,
          'pnl'         : dict — simulated daily P&L per symbol,
          'total_pnl'   : float — sum of all P&L values
        }

    Example
    -------
    >>> report = generate_daily_report(['AAPL', 'SPY'])
    >>> print(report['total_pnl'])
    12.45
    """

    prices = fetch_market_data(symbols)

    # Simulate P&L as a small random percentage of each price
    pnl = {
        symbol: round(price * random.uniform(-0.02, 0.02), 2)
        for symbol, price in prices.items()
    }

    total_pnl = round(sum(pnl.values()), 2)

    report = {
        "date"     : datetime.now().strftime("%Y-%m-%d"),
        "symbols"  : symbols,
        "prices"   : prices,
        "pnl"      : pnl,
        "total_pnl": total_pnl,
    }

    logger.info(
        "generate_daily_report | date=%s total_pnl=%s",
        report["date"], total_pnl
    )
    return report


# Quick test
report = generate_daily_report(["AAPL", "MSFT", "SPY"])
for k, v in report.items():
    print(f"{k:12s}: {v}")

def register_trading_schedule(
    watchlist  : List[str],
    market_open: str = "09:30",
    market_close: str = "16:00",
    data_interval_minutes: int = 5,
) -> None:
    """
    Register all trading tasks with the `schedule` library.

    Sets up three recurring jobs:
      1. Market data fetch on a fixed minute interval throughout the day.
      2. Signal generation at market open.
      3. Daily report generation at market close.

    Parameters
    ----------
    watchlist : list of str
        Symbols to track and trade.
    market_open : str, optional
        Time string for the market-open job in HH:MM format. Default: '09:30'.
    market_close : str, optional
        Time string for the market-close job in HH:MM format. Default: '16:00'.
    data_interval_minutes : int, optional
        How often to fetch market data, in minutes. Default: 5.

    Returns
    -------
    None  (registers jobs as a side-effect)

    Example
    -------
    >>> register_trading_schedule(['AAPL', 'MSFT'], data_interval_minutes=1)
    >>> schedule.run_pending()   # manually trigger any due jobs
    """

    # Clear any previously registered jobs to avoid duplicates on re-run
    schedule.clear()

    # --- Job 1: Fetch market data every N minutes ---
    # Uses a lambda to pass the watchlist argument into the scheduled call
    schedule.every(data_interval_minutes).minutes.do(
        lambda: fetch_market_data(watchlist)
    )

    # --- Job 2: Generate signals at market open ---
    def open_routine():
        prices  = fetch_market_data(watchlist)
        signals = generate_signals(prices)
        execute_orders(signals, dry_run=True)

    schedule.every().day.at(market_open).do(open_routine)

    # --- Job 3: Daily report at market close ---
    schedule.every().day.at(market_close).do(
        lambda: generate_daily_report(watchlist)
    )

    # Print a summary of what was registered
    print(f"Registered {len(schedule.jobs)} jobs:")
    for job in schedule.jobs:
        print(f"  {job}")


register_trading_schedule(
    watchlist             = ["AAPL", "MSFT", "SPY"],
    data_interval_minutes = 1,
)

def run_schedule_loop(duration_seconds: int = 10, poll_interval: float = 1.0) -> None:
    """
    Run the `schedule` event loop for a fixed duration.

    In a real application you would run this loop indefinitely in a
    background thread or a standalone process.  Here we run it for a short
    fixed period so the notebook cell completes.

    Parameters
    ----------
    duration_seconds : int, optional
        How many seconds to keep the loop running. Default: 10.
    poll_interval : float, optional
        Seconds to sleep between each `run_pending()` call. Default: 1.0.

    Returns
    -------
    None

    Example
    -------
    >>> run_schedule_loop(duration_seconds=30)
    """

    end_time = time.time() + duration_seconds
    logger.info("schedule loop starting — will run for %ds", duration_seconds)

    while time.time() < end_time:
        # Run any jobs whose next scheduled time has passed
        schedule.run_pending()
        time.sleep(poll_interval)

    logger.info("schedule loop finished.")


# Force all pending jobs to run once now (overrides their normal timing)
# so we can see output without waiting for the real scheduled times.
schedule.run_all()

print("\nAll registered jobs executed once above via run_all().")

def build_trading_scheduler(
    watchlist   : List[str],
    timezone    : str = "America/New_York",
) -> BackgroundScheduler:
    """
    Build and configure an APScheduler BackgroundScheduler for trading tasks.

    Registers four jobs using real cron expressions:
      1. Pre-market data fetch — every weekday at 08:30
      2. Market-open routine  — every weekday at 09:30
      3. Intraday data poll   — every 5 minutes during trading hours
      4. End-of-day report    — every weekday at 16:05

    Parameters
    ----------
    watchlist : list of str
        Symbols to pass to each job.
    timezone : str, optional
        IANA timezone string for cron trigger evaluation.
        Default: 'America/New_York' (NYSE timezone).

    Returns
    -------
    BackgroundScheduler
        Configured scheduler, not yet started.  Call `.start()` to begin.

    Example
    -------
    >>> scheduler = build_trading_scheduler(['AAPL', 'SPY'])
    >>> scheduler.start()
    >>> # ... later ...
    >>> scheduler.shutdown()
    """

    scheduler = BackgroundScheduler(timezone=timezone)

    # --- Job 1: Pre-market data fetch (weekdays 08:30) ---
    scheduler.add_job(
        func      = fetch_market_data,
        trigger   = CronTrigger(day_of_week="mon-fri", hour=8, minute=30, timezone=timezone),
        args      = [watchlist],
        id        = "pre_market_fetch",
        name      = "Pre-market Data Fetch",
        # If the job was missed (process was down), skip it rather than backfill
        misfire_grace_time = 60,
    )

    # --- Job 2: Market open routine (weekdays 09:30) ---
    def market_open_routine():
        prices  = fetch_market_data(watchlist)
        signals = generate_signals(prices)
        execute_orders(signals, dry_run=True)

    scheduler.add_job(
        func      = market_open_routine,
        trigger   = CronTrigger(day_of_week="mon-fri", hour=9, minute=30, timezone=timezone),
        id        = "market_open",
        name      = "Market Open Routine",
        misfire_grace_time = 60,
    )

    # --- Job 3: Intraday data poll every 5 minutes ---
    # IntervalTrigger is simpler than a cron expression for fixed intervals
    scheduler.add_job(
        func      = fetch_market_data,
        trigger   = IntervalTrigger(minutes=5),
        args      = [watchlist],
        id        = "intraday_poll",
        name      = "Intraday Data Poll",
        misfire_grace_time = 30,
    )

    # --- Job 4: End-of-day report (weekdays 16:05, five minutes after close) ---
    scheduler.add_job(
        func      = generate_daily_report,
        trigger   = CronTrigger(day_of_week="mon-fri", hour=16, minute=5, timezone=timezone),
        args      = [watchlist],
        id        = "eod_report",
        name      = "End-of-Day Report",
        misfire_grace_time = 300,
    )

    logger.info("Scheduler built with %d jobs.", len(scheduler.get_jobs()))
    return scheduler


# Build (but do not start) the scheduler
WATCHLIST = ["AAPL", "MSFT", "GOOGL", "SPY"]
scheduler = build_trading_scheduler(WATCHLIST)

print("\nRegistered jobs:")
for job in scheduler.get_jobs():
    # Safely get next_run_time attribute, defaulting to None if it doesn't exist
    next_run_time = getattr(job, 'next_run_time', None)

    # Format the next_run_time for display
    next_run_str = str(next_run_time)[:19] if next_run_time else "N/A"

    print(f"  {job.id:20s}  {job.name:30s}  next_run: {next_run_str}")

@dataclass
class JobAuditEntry:
    """Records a single management action performed on a scheduled job."""
    timestamp : str
    job_id    : str
    action    : str   # 'add', 'pause', 'resume', 'remove', 'trigger'
    detail    : str


class JobManager:
    """
    A wrapper around APScheduler that adds runtime job management and auditing.

    Provides methods to add, pause, resume, remove, and manually trigger
    jobs without restarting the scheduler.  All operations are logged to
    an in-memory audit trail.

    Parameters
    ----------
    scheduler : BackgroundScheduler
        A configured (but not necessarily started) APScheduler instance.

    Attributes
    ----------
    audit_log : list of JobAuditEntry
        Chronological record of all management actions.

    Example
    -------
    >>> mgr = JobManager(scheduler)
    >>> mgr.pause_job('intraday_poll')
    >>> mgr.resume_job('intraday_poll')
    >>> mgr.print_audit_log()
    """

    def __init__(self, scheduler: BackgroundScheduler):
        self.scheduler  = scheduler
        self.audit_log: List[JobAuditEntry] = []

    def _record(self, job_id: str, action: str, detail: str = "") -> None:
        """Append an entry to the audit log and emit a log line."""
        entry = JobAuditEntry(
            timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
            job_id    = job_id,
            action    = action,
            detail    = detail,
        )
        self.audit_log.append(entry)
        logger.info("JobManager | %s | job_id=%s | %s", action.upper(), job_id, detail)

    def add_one_shot_job(
        self,
        job_id  : str,
        func    : Callable,
        run_at  : datetime,
        args    : Optional[list] = None,
        kwargs  : Optional[dict] = None,
    ) -> None:
        """
        Schedule a function to run once at a specific datetime.

        Useful for handling ad-hoc events such as earnings announcements
        or manual intervention during a market disruption.

        Parameters
        ----------
        job_id  : str      — unique identifier for the job
        func    : callable — function to execute
        run_at  : datetime — exact time to run the job
        args    : list     — positional arguments for func
        kwargs  : dict     — keyword arguments for func

        Example
        -------
        >>> from datetime import datetime, timedelta
        >>> mgr.add_one_shot_job(
        ...     'earnings_check',
        ...     fetch_market_data,
        ...     run_at=datetime.now() + timedelta(seconds=5),
        ...     args=[['AAPL']],
        ... )
        """
        from apscheduler.triggers.date import DateTrigger

        self.scheduler.add_job(
            func    = func,
            trigger = DateTrigger(run_date=run_at),
            id      = job_id,
            args    = args or [],
            kwargs  = kwargs or {},
        )
        self._record(job_id, "add", f"one-shot at {run_at.isoformat()}")

    def pause_job(self, job_id: str) -> None:
        """
        Pause a job so it stops firing without removing it.

        Parameters
        ----------
        job_id : str — ID of the job to pause
        """
        self.scheduler.pause_job(job_id)
        self._record(job_id, "pause")

    def resume_job(self, job_id: str) -> None:
        """
        Resume a previously paused job.

        Parameters
        ----------
        job_id : str — ID of the job to resume
        """
        self.scheduler.resume_job(job_id)
        self._record(job_id, "resume")

    def remove_job(self, job_id: str) -> None:
        """
        Permanently remove a job from the scheduler.

        Parameters
        ----------
        job_id : str — ID of the job to remove
        """
        self.scheduler.remove_job(job_id)
        self._record(job_id, "remove")

    def list_jobs(self) -> None:
        """Print a summary of all currently registered jobs and their status."""
        jobs = self.scheduler.get_jobs()
        print(f"{'ID':<22} {'Name':<32} {'Next Run':<22} {'State'}")
        print("-" * 90)
        for job in jobs:
            # Safely get next_run_time attribute, defaulting to None if it doesn't exist
            next_run_time = getattr(job, 'next_run_time', None)

            state    = "PAUSED" if next_run_time is None else "ACTIVE"
            next_run = str(next_run_time)[:19] if next_run_time else "(paused)"
            print(f"{job.id:<22} {job.name:<32} {next_run:<22} {state}")

    def print_audit_log(self) -> None:
        """Print all recorded management actions in chronological order."""
        print(f"{'Timestamp':<22} {'Job ID':<22} {'Action':<10} Detail")
        print("-" * 80)
        for entry in self.audit_log:
            print(f"{entry.timestamp:<22} {entry.job_id:<22} {entry.action:<10} {entry.detail}")


# Instantiate the manager (scheduler is not started yet)
manager = JobManager(scheduler)
manager.list_jobs()

def run_scheduler_demo(
    manager         : JobManager,
    demo_duration_s : int = 8,
) -> None:
    """
    Start the scheduler, fire all jobs manually for demonstration, then stop.

    This function exercises the full lifecycle in a notebook-friendly way:
    start -> manual trigger -> pause/resume demo -> one-shot job -> shutdown.

    Parameters
    ----------
    manager : JobManager
        The JobManager instance wrapping the configured scheduler.
    demo_duration_s : int, optional
        Seconds to keep the scheduler running after setup. Default: 8.

    Returns
    -------
    None

    Example
    -------
    >>> run_scheduler_demo(manager, demo_duration_s=5)
    """

    sched = manager.scheduler

    # --- Start the background scheduler thread ---
    sched.start()
    logger.info("Scheduler started.")

    # --- Manually fire each job once so output is visible immediately ---
    logger.info("--- Manually triggering all jobs for demonstration ---")
    for job in sched.get_jobs():
        job.func(*job.args, **job.kwargs)

    time.sleep(1)

    # --- Demonstrate pause and resume ---
    logger.info("--- Pausing intraday_poll ---")
    manager.pause_job("intraday_poll")
    time.sleep(1)

    logger.info("--- Resuming intraday_poll ---")
    manager.resume_job("intraday_poll")
    time.sleep(1)

    # --- Add a one-shot job that fires 3 seconds from now ---
    logger.info("--- Scheduling one-shot job in 3s ---")
    manager.add_one_shot_job(
        job_id = "one_shot_check",
        func   = fetch_market_data,
        run_at = datetime.now() + timedelta(seconds=3),
        args   = [["AAPL"]],
    )

    # Wait long enough for the one-shot job to fire
    time.sleep(demo_duration_s)

    # --- Shut down cleanly (waits for running jobs to finish) ---
    sched.shutdown(wait=True)
    logger.info("Scheduler shut down cleanly.")

    print("\n--- Audit Log ---")
    manager.print_audit_log()


run_scheduler_demo(manager, demo_duration_s=6)

def with_retry(
    func          : Callable,
    max_retries   : int = 3,
    retry_delay_s : float = 2.0,
    fallback      : Any = None,
) -> Callable:
    """
    Wrap a callable with retry logic for use in scheduled jobs.

    Returns a new callable that re-invokes `func` up to `max_retries` times
    if an exception is raised, waiting `retry_delay_s` seconds between each
    attempt.  If all retries are exhausted, logs the error and returns
    `fallback` instead of raising, so the scheduler continues running.

    Parameters
    ----------
    func : callable
        The function to wrap.
    max_retries : int, optional
        Maximum number of retry attempts after the first failure. Default: 3.
    retry_delay_s : float, optional
        Seconds to sleep between retries. Default: 2.0.
    fallback : any, optional
        Value to return if all retries fail. Default: None.

    Returns
    -------
    callable
        A wrapped version of func with retry behaviour.

    Example
    -------
    >>> safe_fetch = with_retry(fetch_market_data, max_retries=2)
    >>> prices = safe_fetch(['AAPL'])   # retries up to 2 times on failure
    """

    def wrapper(*args, **kwargs):
        last_exception = None

        for attempt in range(1, max_retries + 2):  # +2 so attempt 1 is the first try
            try:
                return func(*args, **kwargs)

            except Exception as exc:
                last_exception = exc
                logger.warning(
                    "with_retry | %s attempt %d/%d failed: %s",
                    func.__name__, attempt, max_retries + 1, str(exc)
                )

                # Only sleep if there are retries left
                if attempt <= max_retries:
                    time.sleep(retry_delay_s)

        # All retries exhausted
        logger.error(
            "with_retry | %s failed after %d attempts. Last error: %s",
            func.__name__, max_retries + 1, str(last_exception)
        )
        return fallback

    # Preserve the original function's name and docstring
    wrapper.__name__ = func.__name__
    wrapper.__doc__  = func.__doc__
    return wrapper


# --- Demonstrate with a function that fails on purpose ---
def flaky_fetch(symbols: List[str]) -> Dict[str, float]:
    """Simulates a data fetch that fails 70% of the time."""
    if random.random() < 0.7:
        raise ConnectionError("Simulated API timeout")
    return fetch_market_data(symbols)


safe_fetch = with_retry(flaky_fetch, max_retries=4, retry_delay_s=0.1)

result = safe_fetch(["AAPL", "MSFT"])
print(f"\nResult after retries: {result}")

def describe_cron_expression(expression: str) -> str:
    """
    Return a human-readable description of a cron expression.

    Covers the most common trading-relevant patterns. For expressions not
    in the lookup table, returns the raw expression with field labels.

    Parameters
    ----------
    expression : str
        A standard five-field cron string, e.g. '0 9 * * 1-5'.

    Returns
    -------
    str
        Human-readable description.

    Example
    -------
    >>> describe_cron_expression('0 9 * * 1-5')
    'Every weekday at 09:00'
    """

    known_patterns = {
        "* * * * *"      : "Every minute",
        "*/5 * * * *"    : "Every 5 minutes",
        "*/15 * * * *"   : "Every 15 minutes",
        "0 * * * *"      : "Every hour at the top of the hour",
        "0 9 * * 1-5"    : "Every weekday at 09:00",
        "30 8 * * 1-5"   : "Every weekday at 08:30 (pre-market)",
        "30 9 * * 1-5"   : "Every weekday at 09:30 (NYSE open)",
        "0 16 * * 1-5"   : "Every weekday at 16:00 (NYSE close)",
        "5 16 * * 1-5"   : "Every weekday at 16:05 (post-market)",
        "0 0 * * *"      : "Every day at midnight",
        "0 0 * * 1"      : "Every Monday at midnight (weekly)",
        "0 0 1 * *"      : "First day of every month at midnight",
        "0 0 1 1 *"      : "Annually on January 1st",
    }

    if expression in known_patterns:
        return known_patterns[expression]

    # Fall back to a labelled field breakdown
    fields = expression.split()
    if len(fields) != 5:
        return f"Invalid cron expression (expected 5 fields, got {len(fields)})"

    labels = ["minute", "hour", "day-of-month", "month", "day-of-week"]
    return "Cron: " + ", ".join(f"{l}={v}" for l, v in zip(labels, fields))


def preview_next_runs(
    expression : str,
    count      : int = 5,
    timezone   : str = "America/New_York",
) -> List[str]:
    """
    Compute the next N fire times for a cron expression.

    Uses APScheduler's CronTrigger to calculate exact datetimes, respecting
    the given timezone.

    Parameters
    ----------
    expression : str  — five-field cron string
    count      : int  — number of future fire times to return. Default: 5.
    timezone   : str  — IANA timezone for calculations. Default: 'America/New_York'.

    Returns
    -------
    list of str
        ISO-format datetime strings for the next `count` fire times.

    Example
    -------
    >>> preview_next_runs('30 9 * * 1-5', count=3)
    ['2024-06-10 09:30:00-04:00', '2024-06-11 09:30:00-04:00', ...]
    """

    minute, hour, day, month, dow = expression.split()
    trigger  = CronTrigger(
        minute      = minute,
        hour        = hour,
        day         = day,
        month       = month,
        day_of_week = dow,
        timezone    = timezone,
    )

    fire_times = []
    current    = datetime.now()

    for _ in range(count):
        next_time = trigger.get_next_fire_time(current, current)
        if next_time is None:
            break
        fire_times.append(str(next_time)[:19])
        # Advance by one second past the last fire time to get the next one
        current = next_time + timedelta(seconds=1)

    return fire_times


# --- Demo: describe and preview common trading cron expressions ---
trading_crons = [
    "30 8 * * 1-5",
    "30 9 * * 1-5",
    "*/5 * * * *",
    "5 16 * * 1-5",
    "0 0 * * 1",
]

for expr in trading_crons:
    desc      = describe_cron_expression(expr)
    next_runs = preview_next_runs(expr, count=2)
    print(f"  {expr:<20} {desc}")
    print(f"  {'':20} Next: {next_runs}")
    print()

class TradingSchedulerPipeline:
    """
    A self-contained scheduler pipeline for automated trading tasks.

    Combines APScheduler job management, retry-wrapped task functions, and
    an audit log into a single object that can be started and stopped as a
    unit. Designed to run as a long-lived background process.

    Parameters
    ----------
    watchlist : list of str
        Ticker symbols to monitor and trade.
    timezone : str, optional
        IANA timezone for all cron trigger evaluations.
        Default: 'America/New_York'.
    dry_run : bool, optional
        When True, order execution is simulated. Default: True.

    Attributes
    ----------
    manager : JobManager
        Exposes pause, resume, remove, and add_one_shot_job methods.

    Example
    -------
    >>> pipeline = TradingSchedulerPipeline(
    ...     watchlist = ['AAPL', 'MSFT', 'SPY'],
    ...     dry_run   = True,
    ... )
    >>> pipeline.start()
    >>> pipeline.status()
    >>> pipeline.stop()
    """

    def __init__(
        self,
        watchlist : List[str],
        timezone  : str  = "America/New_York",
        dry_run   : bool = True,
    ):
        self.watchlist = watchlist
        self.timezone  = timezone
        self.dry_run   = dry_run

        # Wrap task functions with retry logic
        self._fetch   = with_retry(fetch_market_data, max_retries=3, retry_delay_s=5)
        self._report  = with_retry(generate_daily_report, max_retries=2, retry_delay_s=10)

        # Build scheduler and wrap it in the manager
        _scheduler    = BackgroundScheduler(timezone=timezone)
        self.manager  = JobManager(_scheduler)

        self._register_jobs()

    def _register_jobs(self) -> None:
        """Register all trading jobs on the internal scheduler."""

        sched = self.manager.scheduler

        # Pre-market fetch
        sched.add_job(
            func    = self._fetch,
            trigger = CronTrigger(day_of_week="mon-fri", hour=8, minute=30),
            args    = [self.watchlist],
            id      = "pre_market_fetch",
            name    = "Pre-market Data Fetch",
        )

        # Market-open pipeline
        sched.add_job(
            func    = self._run_open_pipeline,
            trigger = CronTrigger(day_of_week="mon-fri", hour=9, minute=30),
            id      = "market_open",
            name    = "Market Open Pipeline",
        )

        # Intraday poll every 5 minutes
        sched.add_job(
            func    = self._fetch,
            trigger = IntervalTrigger(minutes=5),
            args    = [self.watchlist],
            id      = "intraday_poll",
            name    = "Intraday Data Poll",
        )

        # End-of-day report
        sched.add_job(
            func    = self._report,
            trigger = CronTrigger(day_of_week="mon-fri", hour=16, minute=5),
            args    = [self.watchlist],
            id      = "eod_report",
            name    = "End-of-Day Report",
        )

    def _run_open_pipeline(self) -> None:
        """Execute the full market-open sequence: fetch -> signal -> order."""
        prices  = self._fetch(self.watchlist)
        if prices is None:
            logger.error("Market open pipeline: data fetch failed, skipping signals.")
            return
        signals = generate_signals(prices)
        execute_orders(signals, dry_run=self.dry_run)

    def start(self) -> None:
        """Start the background scheduler. Non-blocking."""
        self.manager.scheduler.start()
        logger.info("TradingSchedulerPipeline started with %d jobs.", len(self.manager.scheduler.get_jobs()))

    def stop(self) -> None:
        """Shut down the scheduler, waiting for any running job to finish."""
        self.manager.scheduler.shutdown(wait=True)
        logger.info("TradingSchedulerPipeline stopped.")

    def status(self) -> None:
        """Print current job status and audit log."""
        print("\n--- Job Status ---")
        self.manager.list_jobs()
        if self.manager.audit_log:
            print("\n--- Audit Log ---")
            self.manager.print_audit_log()


# --- Demo run ---
pipeline = TradingSchedulerPipeline(
    watchlist = ["AAPL", "MSFT", "GOOGL", "SPY"],
    dry_run   = True,
)

pipeline.start()
pipeline.status()

# Manually trigger the open pipeline to see output
logger.info("--- Manually triggering market open pipeline ---")
pipeline._run_open_pipeline()

time.sleep(1)
pipeline.stop()

print("\nPipeline demo complete.")