Learn algo trading one clear chapter at a time.

from dataclasses import dataclass

@dataclass
class StrategySettings:
    symbol: str
    timeframe: str
    max_positions: int
    paper_mode: bool = True

settings = StrategySettings(
    symbol="YOUR_SYMBOL",
    timeframe="5minute",
    max_positions=1,
)

print(f"Starting in paper mode: {settings.paper_mode}")

import os
from dataclasses import dataclass

@dataclass(frozen=True)
class Settings:
    paper_mode: bool = os.getenv("PAPER_MODE", "true").lower() == "true"
    max_order_quantity: int = int(os.getenv("MAX_ORDER_QUANTITY", "1"))

settings = Settings()

if settings.max_order_quantity < 1:
    raise ValueError("MAX_ORDER_QUANTITY must be at least 1")

import pandas as pd

REQUIRED = {"timestamp", "open", "high", "low", "close", "volume"}

def validate_candles(frame: pd.DataFrame) -> pd.DataFrame:
    missing = REQUIRED.difference(frame.columns)
    if missing:
        raise ValueError(f"Missing candle columns: {sorted(missing)}")
    clean = frame.drop_duplicates(subset="timestamp").copy()
    clean["timestamp"] = pd.to_datetime(clean["timestamp"], utc=True)
    clean = clean.sort_values("timestamp").reset_index(drop=True)
    if (clean["high"] < clean[["open", "close", "low"]].max(axis=1)).any():
        raise ValueError("Found an invalid candle high")
    return clean

import pandas as pd

def moving_average_signal(close: pd.Series) -> pd.Series:
    fast = close.rolling(window=10, min_periods=10).mean()
    slow = close.rolling(window=30, min_periods=30).mean()
    signal = pd.Series(0, index=close.index, dtype="int8")
    signal[fast > slow] = 1
    signal[fast < slow] = -1
    return signal

import pandas as pd

def strategy_returns(close: pd.Series, position: pd.Series, fee: float) -> pd.Series:
    market_returns = close.pct_change().fillna(0)
    turnover = position.diff().abs().fillna(position.abs())
    costs = turnover * fee
    return position.shift(1).fillna(0) * market_returns - costs

# Historical estimates depend on data and execution assumptions.
# They do not predict future performance.

def quantity_from_risk(
    account_value: float,
    risk_fraction: float,
    entry_price: float,
    stop_price: float,
    max_quantity: int,
) -> int:
    risk_per_unit = abs(entry_price - stop_price)
    if risk_per_unit <= 0 or max_quantity < 1:
        return 0
    risk_budget = account_value * risk_fraction
    estimated = int(risk_budget // risk_per_unit)
    return max(0, min(estimated, max_quantity))

from dataclasses import dataclass

@dataclass(frozen=True)
class OptionContract:
    symbol: str
    expiry: str
    strike: float
    option_type: str
    lot_size: int

def validate_contract(contract: OptionContract) -> None:
    if contract.option_type not in {"CE", "PE"}:
        raise ValueError("option_type must be CE or PE")
    if contract.strike <= 0 or contract.lot_size <= 0:
        raise ValueError("strike and lot_size must be positive")

import os

def require_api_setting(name: str) -> str:
    value = os.getenv(name)
    if not value:
        raise RuntimeError(f"Set {name} in your secure environment")
    return value

# Example names only. Do not commit actual credentials.
client_id = require_api_setting("BROKER_CLIENT_ID")
api_key = require_api_setting("BROKER_API_KEY")

from enum import Enum

class OrderState(str, Enum):
    CREATED = "created"
    SUBMITTED = "submitted"
    OPEN = "open"
    PARTIAL = "partial"
    COMPLETE = "complete"
    REJECTED = "rejected"
    CANCELLED = "cancelled"

FINAL_STATES = {OrderState.COMPLETE, OrderState.REJECTED, OrderState.CANCELLED}

def preflight(settings: dict) -> list[str]:
    problems = []
    if settings.get("paper_mode") is not True:
        problems.append("Paper mode is not enabled")
    if settings.get("max_order_quantity", 0) < 1:
        problems.append("Maximum quantity must be configured")
    if not settings.get("symbol"):
        problems.append("Instrument must be selected")
    return problems

issues = preflight({"paper_mode": True, "max_order_quantity": 1, "symbol": "DEMO"})
print("Ready for controlled testing" if not issues else issues)

from datetime import datetime, timezone

def data_is_fresh(last_update: datetime, max_age_seconds: int) -> bool:
    now = datetime.now(timezone.utc)
    if last_update.tzinfo is None:
        raise ValueError("last_update must include a timezone")
    age = (now - last_update).total_seconds()
    return 0 <= age <= max_age_seconds

# Never log API keys, passwords, OTPs or session tokens.

from dataclasses import dataclass

@dataclass(frozen=True)
class Preflight:
    data_is_fresh: bool
    risk_limit_ok: bool
    paper_mode: bool = True

def decide_next_step(checks: Preflight) -> str:
    if not checks.data_is_fresh:
        return "Stop: market data is stale"
    if not checks.risk_limit_ok:
        return "Stop: risk check failed"
    if checks.paper_mode:
        return "Continue with paper simulation"
    return "Live execution is disabled in this learning example"

print(decide_next_step(Preflight(True, True)))