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).

quantra-go-algo

def max_drawdown(equity: pd.Series) -> float:
    peak = equity.cummax()
    dd = (equity / peak) - 1.0
    return float(dd.min())


def qs_metrics(equity: pd.Series) -> dict:
    """
    Quantstats-style metrics from an equity curve.
    """

quantra-go-algo

def make_positions(df_feat: pd.DataFrame, regime: pd.Series, params: dict) -> pd.Series:
    """
    Same logic as before, with parameters chosen by WFO.
    """
    z_thr = params["z_thr"]
    range_size = params["range_size"]
    lowvol_size = params["lowvol_size"]
    highvol_size = params["highvol_size"]

    pos = pd.Series(0.0, index=df_feat.index)

quantra-go-algo

def load_cache() -> dict:
    p = Path(LLM_CACHE_FILE)
    if p.exists():
        return json.loads(p.read_text(encoding="utf-8"))
    return {}


def save_cache(cache: dict) -> None:
    Path(LLM_CACHE_FILE).write_text(json.dumps(cache, indent=2, sort_keys=True), encoding="utf-8")

quantra-go-algo

def fit_kmeans_pre_oos(df_feat: pd.DataFrame) -> tuple[KMeans, np.ndarray, np.ndarray, dict[int, str]]:
    """
    Fit KMeans using ONLY data BEFORE OOS_START (prevents training leakage).
    """
    split = pd.to_datetime(OOS_START)

    rows = []
    for t in label_dates(df_feat.index):
        if t >= split:
            break

quantra-go-algo

def load_data() -> pd.DataFrame:
    df = yf.download(SYMBOL, start=START, end=END, interval="1d", progress=False, group_by='tickers')[SYMBOL]
    if df.empty:
        raise RuntimeError(f"No data returned for {SYMBOL}. Check symbol/date range.")
    df = df.rename(columns=str.title)
    if "Close" not in df.columns and "Adj Close" in df.columns:
        df["Close"] = df["Adj Close"]
    df = df.dropna(subset=["Close"])
    df.index = pd.to_datetime(df.index)
    return df

quantra-go-algo

DEEPSEEK_API_KEY = "PASTE_YOUR_DEEPSEEK_KEY_HERE"
DEEPSEEK_BASE_URL = "https://api.deepseek.com"
DEEPSEEK_MODEL = "deepseek-chat"

# --- Market / data ---
SYMBOL = "EURUSD=X"
START = "2006-01-01"
END = "2026-04-11"
OOS_START = "2023-01-01"          # we show equity curves from this date onward

quantra-go-algo

import json
import math
import time
import re
from pathlib import Path

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import yfinance as yf

quantra-go-algo

--- Execution Log ---
--- Starting analysis for AAPL ---
Preliminary decision for AAPL: Signal=0
--- Starting analysis for MSFT ---
Preliminary decision for MSFT: Signal=0
--- Starting analysis for GOOG ---
Preliminary decision for GOOG: Signal=0
Portfolio Agent: No BUY signals. Allocating 0% to all tickers.
Execution Agent: Submitting real trades based on portfolio plan.
Execution: Fetched cash balance: $100000.00