quantra-go-algo

# Retry once on parse failure, fall back to rule policy if both fail
for attempt in range(2):
    content = deepseek_chat(messages, max_tokens=2000)
    try:
        obj = parse_json(content)
        return obj.get('policy', {})
    except ValueError as e:
        if attempt == 1:
            print(f'Parse failed twice for {month_start.date()}, using rule fallback')
            return rule_policy(stats)   # deterministic fallback

quantra-go-algo

system = (
    'You are a risk manager for a long-only AAPL equity strategy. '
    'Your job is NOT to predict tomorrow\'s price direction. '
    'Your job is to assess whether each market state carries elevated '
    'downside risk that warrants reducing exposure. '
    'For each market state, output LONG (stay invested) or FLAT (reduce). '
    'Use size 1.0 for clearly positive or neutral risk/return states. '
    'Use size 0.5 for mildly positive but uncertain states. '
    'FLAT = clearly negative expected return or high downside risk: '
    'use it sparingly, only when evidence is unambiguous. '

quantra-go-algo

def state_stats(train_df):
    tmp = train_df.copy()
    tmp['state_lag'] = tmp['state'].shift(1)   # use yesterday's state
    tmp['ret_fwd']   = tmp['ret']              # to predict today's return
    tmp = tmp.dropna(subset=['state_lag', 'ret_fwd'])

    g = tmp.groupby('state_lag')['ret_fwd']
    stats = pd.DataFrame({
        'count':      g.size(),
        'mean':       g.mean(),

quantra-go-algo

# Trend: above/below zero
trend_bucket = np.where(feat['trend'] >= 0, 'TREND_UP', 'TREND_DOWN')

# Volatility: above/below rolling median (adaptive threshold)
vol_med    = feat['vol'].rolling(252, min_periods=60).median()
vol_bucket = np.where(feat['vol'] > vol_med, 'VOL_HIGH', 'VOL_LOW')

# Z-score: three buckets: oversold / neutral / overbought
z = feat['z']
z_bucket = np.where(z <= -Z_EDGE, 'Z_LOW',

quantra-go-algo

feat = df.copy()

# 1. Daily log returns: the foundation of everything
feat['ret'] = np.log(feat['Close']).diff()

# 2. Realized annualized volatility (20-day rolling)
feat['vol'] = feat['ret'].rolling(VOL_WIN).std() * np.sqrt(252)

# 3. Trend score: rolling mean / rolling std
#    Positive = uptrend, negative = downtrend, magnitude = consistency

quantra-go-algo

# Sanity check: the biggest daily move on split-adjusted AAPL should be
# around -52% (Sep 2000 earnings warning). If you see moves > 100%,
# the data is unadjusted.
_top = feat['ret'].abs().sort_values(ascending=False).head(5)
if feat['ret'].abs().max() > 1.0:
    print('WARNING: possible split artifact: check auto_adjust=True')

quantra-go-algo

import yfinance as yf
import pandas as pd, numpy as np

# auto_adjust=True ensures prices are split- and dividend-adjusted.
# AAPL split 7-for-1 in 2014 and 4-for-1 in 2020. Without adjustment,
# those days show fake -86% and -75% returns that corrupt every feature.
df = yf.download(SYMBOL, start=START, auto_adjust=True, progress=False)

if isinstance(df.columns, pd.MultiIndex):
    df = df.droplevel(-1, axis=1)   # flatten multi-level columns

quantra-go-algo

# ---------- Policy mode ----------
POLICY_MODE   = 'llm'    # 'llm' = DeepSeek  |  'rule' = deterministic baseline

# ---------- Data ----------
SYMBOL        = 'AAPL'
START         = '1990-01-01'
OOS_START     = '2023-01-01'

# ---------- Costs ----------
COST_BPS      = 0.28      # IBKR Fixed: $0.005/share + reg fees at AAPL ~$185

quantra-go-algo

def main():
    if DEEPSEEK_API_KEY.strip() == "PASTE_YOUR_DEEPSEEK_KEY_HERE":
        print("WARNING: You have not set your DeepSeek API key. LLM curve will fail.")
        print("Edit DEEPSEEK_API_KEY at the top of this script.\n")

    print(f"Loading {SYMBOL} from {START} to today ...")
    df = load_data()
    df_feat = add_features(df)

    oos_start = pd.to_datetime(OOS_START)

quantra-go-algo

def month_starts(index: pd.DatetimeIndex, start: pd.Timestamp) -> list[pd.Timestamp]:
    """All month starts >= start that exist in the index."""
    months = pd.date_range(start=start, end=index.max(), freq="MS")
    # Keep only months that have data
    return [m for m in months if (index >= m).any()]


def optimize_params(train_feat: pd.DataFrame, train_regime: pd.Series) -> dict:
    """
    Grid search parameters on training set. Objective: maximize Sharpe (simple & common).