Spaces:

getpulsar
/

signals

Running

App Files Files Community

signals / backend /backtest /engine.py

hardik1731

Claude Sonnet 5

Add Pulsar Signals: automated crypto scanner + quality-gated notifications

da51398 4 months ago

Raw History Blame Contribute Delete

11.1 kB

"""
Backtesting engine — replays the signal logic across historical candles
and measures how often the predicted direction matched actual price movement.
Two candidate models are fit on a chronological split (features = indicator
votes, label = whether price rose over the next `lookahead` candles):
  - LinearModel: logistic regression coefficients used as vote weights
  - TreeModel: gradient boosting (nonlinear, captures indicator interactions)
Whichever validates best is only adopted if it also beats the static
`config.INDICATOR_WEIGHTS` baseline on a final holdout that was never
touched during fitting — otherwise this falls back to the static baseline,
so it can never make accuracy worse than before.
No lookahead bias: indicators are computed rolling from the left, so
df.iloc[i] only reflects data up to candle i. News sentiment is excluded
(no historical news available).
"""
from __future__ import annotations

from typing import Dict, List, Tuple

import numpy as np
import pandas as pd

from signals.scorer import LinearModel, ScoreModel, TreeModel, compute_signal, compute_votes

try:
    from sklearn.ensemble import HistGradientBoostingClassifier
    from sklearn.linear_model import LogisticRegression
    _SKLEARN_AVAILABLE = True
except ImportError:
    _SKLEARN_AVAILABLE = False

MIN_TRAIN_ROWS = 60
MIN_TEST_ROWS = 20
CANDIDATE_THRESHOLDS = [0.0, 0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.45, 0.5, 0.6]
# Low floor lets the threshold search trade coverage for precision (fewer,
# higher-conviction signals); the absolute minimum below still guards against
# "100% accuracy from 2 lucky signals".
MIN_SIGNAL_RATE = 0.08
MIN_THRESHOLD_SIGNALS = 8
MIN_VAL_SIGNALS = 5
MIN_HOLDOUT_SIGNALS = 10  # a "win" on fewer signals than this isn't trustworthy

# Training-label deadband: moves smaller than this many ATRs over the
# lookahead window are direction coin-flips — excluded from training so the
# models learn from meaningful moves only. Backtest scoring is NOT filtered.
LABEL_DEADBAND_ATR = 0.15

# L2 strength for logistic regression: smaller = more regularized.
# Kept small — each extra candidate is another full model fit, and fitting
# runs synchronously per request (see main.py's note on GIL/BLAS contention
# making concurrent fits slower, not faster, on typical hardware).
CANDIDATE_C = [0.05, 0.25, 1.0]

# One conservative gradient-boosting config (small ensemble, shallow trees) —
# more aggressive ones overfit badly on the few hundred rows available per
# symbol, and each additional candidate adds real per-request latency.
CANDIDATE_GBM_PARAMS = [
    {"max_iter": 60, "max_depth": 2, "learning_rate": 0.08, "l2_regularization": 1.5},
]


def _build_training_rows(
    df: pd.DataFrame, lookahead: int
) -> Tuple[List[Dict[str, float]], np.ndarray, List[str]]:
    """Returns (vote_maps, labels, keys) for every row with a known outcome."""
    closes = df["close"].to_numpy()
    atrs = df["ATR_14"].to_numpy() if "ATR_14" in df.columns else None
    usable = len(df) - lookahead

    keys: List[str] = []
    vote_maps: List[Dict[str, float]] = []
    labels: List[int] = []

    for i in range(usable):
        votes = compute_votes(df.iloc[i])
        vote_map = {k: v for k, v, _ in votes if k != "news_sentiment"}
        if not vote_map:
            continue
        future, current = closes[i + lookahead], closes[i]
        move = future - current
        if move == 0:
            continue
        if (
            atrs is not None
            and not np.isnan(atrs[i])
            and abs(move) < LABEL_DEADBAND_ATR * atrs[i]
        ):
            continue
        for k in vote_map:
            if k not in keys:
                keys.append(k)
        vote_maps.append(vote_map)
        labels.append(1 if move > 0 else 0)

    return vote_maps, np.array(labels), keys


def _score_rows(vote_maps: List[Dict[str, float]], model: ScoreModel) -> np.ndarray:
    return np.array([model.score(vm) for vm in vote_maps])


def _best_threshold(scores: np.ndarray, y: np.ndarray) -> float:
    best_t, best_acc = 0.15, -1.0
    n = len(y)
    for t in CANDIDATE_THRESHOLDS:
        mask = np.abs(scores) > t
        if mask.sum() < max(MIN_THRESHOLD_SIGNALS, n * MIN_SIGNAL_RATE):
            continue
        pred = (scores[mask] > 0).astype(int)
        acc = (pred == y[mask]).mean()
        if acc > best_acc:
            best_acc, best_t = acc, t
    return best_t


def _accuracy_on(vote_maps, y, model: ScoreModel, threshold) -> Tuple[float, int]:
    scores = _score_rows(vote_maps, model)
    mask = np.abs(scores) > threshold
    if mask.sum() == 0:
        return 0.0, 0
    pred = (scores[mask] > 0).astype(int)
    return float((pred == y[mask]).mean()), int(mask.sum())


def _candidate_models(keys: List[str], allow_tree: bool = True):
    """Yields (name, fit_fn) pairs. fit_fn(X, y) -> ScoreModel."""
    for c in CANDIDATE_C:
        def fit_fn(X, y, c=c):
            clf = LogisticRegression(max_iter=1000, C=c)
            clf.fit(X, y)
            return LinearModel({k: float(coef) for k, coef in zip(keys, clf.coef_[0])})
        yield f"logistic_regression(C={c})", fit_fn

    if not allow_tree:
        return

    for params in CANDIDATE_GBM_PARAMS:
        def fit_fn(X, y, params=params):
            clf = HistGradientBoostingClassifier(random_state=42, **params)
            clf.fit(X, y)
            return TreeModel(clf, keys)
        yield f"gradient_boosting({params})", fit_fn


def fit_model(
    df: pd.DataFrame, lookahead: int, base_weights: Dict[str, float], allow_tree: bool = True
) -> Tuple[ScoreModel, float, Dict]:
    """Fits several candidate models (logistic regression at various
    regularization strengths, gradient boosting at a few conservative
    configs), but only adopts one if it actually beats the static
    `base_weights` on a final holdout that was never touched during
    fitting — otherwise falls back to `base_weights`, so this can never
    make accuracy worse than the static baseline.
    Layout (chronological, oldest to newest):
      [ inner-train | inner-val | final holdout ]
      inner-train/inner-val pick the best candidate model and score
      threshold; the final holdout is only ever scored once, to decide
      whether to keep the fitted model or the static one, and to report
      the accuracy shown to the user.
    Falls back to `LinearModel(base_weights)` and threshold 0.15 when there
    isn't enough history yet (new listings, short candle windows) or
    scikit-learn isn't installed. The returned diagnostics include
    `split_index`: callers should backtest only rows from `split_index`
    onward (the final holdout) to reproduce the same accuracy figure.
    `allow_tree=False` skips the gradient-boosting candidate entirely,
    leaving only logistic regression — used by the signals scanner
    (backend/scan/engine.py), which needs a JSON-serializable weights dict
    to cache in the database rather than a pickled sklearn object.
    """
    fallback = (LinearModel(base_weights), 0.15, {"method": "static", "split_index": 0})
    if not _SKLEARN_AVAILABLE:
        return fallback

    vote_maps, y, keys = _build_training_rows(df, lookahead)
    n = len(vote_maps)
    if n < MIN_TRAIN_ROWS + MIN_TEST_ROWS or len(np.unique(y)) < 2:
        return fallback

    holdout_start = int(n * 0.7)
    inner_val_start = int(holdout_start * 0.7)
    if len(np.unique(y[:inner_val_start])) < 2 or (holdout_start - inner_val_start) < 15:
        return fallback

    X = np.array([[vm.get(k, 0.0) for k in keys] for vm in vote_maps])

    inner_train_vm, inner_val_vm = vote_maps[:inner_val_start], vote_maps[inner_val_start:holdout_start]
    inner_train_y, inner_val_y = y[:inner_val_start], y[inner_val_start:holdout_start]

    best_name, best_fit_fn, best_val_acc = None, None, -1.0
    for name, fit_fn in _candidate_models(keys, allow_tree):
        model = fit_fn(X[:inner_val_start], inner_train_y)
        train_scores = _score_rows(inner_train_vm, model)
        t = _best_threshold(train_scores, inner_train_y)
        val_acc, val_signals = _accuracy_on(inner_val_vm, inner_val_y, model, t)
        if val_signals >= MIN_VAL_SIGNALS and val_acc > best_val_acc:
            best_val_acc, best_name, best_fit_fn = val_acc, name, fit_fn

    if best_fit_fn is None:
        return fallback

    # Refit on the full pre-holdout region (inner-train + inner-val) with the
    # winning candidate's config, using all data available before the
    # untouched final holdout.
    fitted_model = best_fit_fn(X[:holdout_start], y[:holdout_start])
    fit_vm, fit_y = vote_maps[:holdout_start], y[:holdout_start]
    fitted_scores = _score_rows(fit_vm, fitted_model)
    fitted_threshold = _best_threshold(fitted_scores, fit_y)

    holdout_vm, holdout_y = vote_maps[holdout_start:], y[holdout_start:]
    fitted_acc, fitted_signals = _accuracy_on(holdout_vm, holdout_y, fitted_model, fitted_threshold)
    static_model = LinearModel(base_weights)
    static_acc, static_signals = _accuracy_on(holdout_vm, holdout_y, static_model, 0.15)

    diagnostics = {
        "split_index": holdout_start,
        "candidate": best_name,
        "fitted_holdout_accuracy": round(fitted_acc * 100, 1),
        "fitted_holdout_signals": fitted_signals,
        "static_holdout_accuracy": round(static_acc * 100, 1),
        "static_holdout_signals": static_signals,
    }

    if fitted_signals >= MIN_HOLDOUT_SIGNALS and fitted_acc > static_acc:
        diagnostics["method"] = "gradient_boosting" if isinstance(fitted_model, TreeModel) else "logistic_regression"
        return fitted_model, fitted_threshold, diagnostics

    diagnostics["method"] = "static"
    diagnostics["reason"] = "fitted model did not beat static baseline out-of-sample"
    return static_model, 0.15, diagnostics


def run_backtest(
    df: pd.DataFrame,
    model,
    lookahead: int,
    threshold: float = 0.15,
    start_index: int = 0,
) -> Dict:
    """Backtests `model`/`threshold` from `start_index` onward.
    Pass `start_index=diagnostics["split_index"]` from `fit_model` to
    measure accuracy only on the held-out test portion (out-of-sample).
    Defaults to 0 (whole-history backtest) for static/fallback weights.
    `model` may be a plain weights dict or any ScoreModel (see scorer.py).
    """
    closes = df["close"].to_numpy()
    n = len(df)
    correct = 0
    total = 0

    for i in range(start_index, n - lookahead):
        row = df.iloc[i]
        score, signal, _ = compute_signal(row, model, news_score=None, threshold=threshold)
        if signal == "NEUTRAL":
            continue

        current = closes[i]
        future = closes[i + lookahead]

        if signal == "BULLISH" and future > current:
            correct += 1
        elif signal == "BEARISH" and future < current:
            correct += 1
        total += 1

    accuracy = round(correct / total * 100, 1) if total > 0 else 0.0
    return {"accuracy": accuracy, "total_signals": total, "correct_signals": correct}

	"""
	Backtesting engine — replays the signal logic across historical candles
	and measures how often the predicted direction matched actual price movement.

	Two candidate models are fit on a chronological split (features = indicator
	votes, label = whether price rose over the next `lookahead` candles):
	- LinearModel: logistic regression coefficients used as vote weights
	- TreeModel: gradient boosting (nonlinear, captures indicator interactions)
	Whichever validates best is only adopted if it also beats the static
	`config.INDICATOR_WEIGHTS` baseline on a final holdout that was never
	touched during fitting — otherwise this falls back to the static baseline,
	so it can never make accuracy worse than before.

	No lookahead bias: indicators are computed rolling from the left, so
	df.iloc[i] only reflects data up to candle i. News sentiment is excluded
	(no historical news available).
	"""
	from __future__ import annotations

	from typing import Dict, List, Tuple

	import numpy as np
	import pandas as pd

	from signals.scorer import LinearModel, ScoreModel, TreeModel, compute_signal, compute_votes

	try:
	from sklearn.ensemble import HistGradientBoostingClassifier
	from sklearn.linear_model import LogisticRegression
	_SKLEARN_AVAILABLE = True
	except ImportError:
	_SKLEARN_AVAILABLE = False

	MIN_TRAIN_ROWS = 60
	MIN_TEST_ROWS = 20
	CANDIDATE_THRESHOLDS = [0.0, 0.05, 0.1, 0.15, 0.2, 0.25, 0.3, 0.35, 0.4, 0.45, 0.5, 0.6]
	# Low floor lets the threshold search trade coverage for precision (fewer,
	# higher-conviction signals); the absolute minimum below still guards against
	# "100% accuracy from 2 lucky signals".
	MIN_SIGNAL_RATE = 0.08
	MIN_THRESHOLD_SIGNALS = 8
	MIN_VAL_SIGNALS = 5
	MIN_HOLDOUT_SIGNALS = 10 # a "win" on fewer signals than this isn't trustworthy

	# Training-label deadband: moves smaller than this many ATRs over the
	# lookahead window are direction coin-flips — excluded from training so the
	# models learn from meaningful moves only. Backtest scoring is NOT filtered.
	LABEL_DEADBAND_ATR = 0.15

	# L2 strength for logistic regression: smaller = more regularized.
	# Kept small — each extra candidate is another full model fit, and fitting
	# runs synchronously per request (see main.py's note on GIL/BLAS contention
	# making concurrent fits slower, not faster, on typical hardware).
	CANDIDATE_C = [0.05, 0.25, 1.0]

	# One conservative gradient-boosting config (small ensemble, shallow trees) —
	# more aggressive ones overfit badly on the few hundred rows available per
	# symbol, and each additional candidate adds real per-request latency.
	CANDIDATE_GBM_PARAMS = [
	{"max_iter": 60, "max_depth": 2, "learning_rate": 0.08, "l2_regularization": 1.5},
	]


	def _build_training_rows(
	df: pd.DataFrame, lookahead: int
	) -> Tuple[List[Dict[str, float]], np.ndarray, List[str]]:
	"""Returns (vote_maps, labels, keys) for every row with a known outcome."""
	closes = df["close"].to_numpy()
	atrs = df["ATR_14"].to_numpy() if "ATR_14" in df.columns else None
	usable = len(df) - lookahead

	keys: List[str] = []
	vote_maps: List[Dict[str, float]] = []
	labels: List[int] = []

	for i in range(usable):
	votes = compute_votes(df.iloc[i])
	vote_map = {k: v for k, v, _ in votes if k != "news_sentiment"}
	if not vote_map:
	continue
	future, current = closes[i + lookahead], closes[i]
	move = future - current
	if move == 0:
	continue
	if (
	atrs is not None
	and not np.isnan(atrs[i])
	and abs(move) < LABEL_DEADBAND_ATR * atrs[i]
	):
	continue
	for k in vote_map:
	if k not in keys:
	keys.append(k)
	vote_maps.append(vote_map)
	labels.append(1 if move > 0 else 0)

	return vote_maps, np.array(labels), keys


	def _score_rows(vote_maps: List[Dict[str, float]], model: ScoreModel) -> np.ndarray:
	return np.array([model.score(vm) for vm in vote_maps])


	def _best_threshold(scores: np.ndarray, y: np.ndarray) -> float:
	best_t, best_acc = 0.15, -1.0
	n = len(y)
	for t in CANDIDATE_THRESHOLDS:
	mask = np.abs(scores) > t
	if mask.sum() < max(MIN_THRESHOLD_SIGNALS, n * MIN_SIGNAL_RATE):
	continue
	pred = (scores[mask] > 0).astype(int)
	acc = (pred == y[mask]).mean()
	if acc > best_acc:
	best_acc, best_t = acc, t
	return best_t


	def _accuracy_on(vote_maps, y, model: ScoreModel, threshold) -> Tuple[float, int]:
	scores = _score_rows(vote_maps, model)
	mask = np.abs(scores) > threshold
	if mask.sum() == 0:
	return 0.0, 0
	pred = (scores[mask] > 0).astype(int)
	return float((pred == y[mask]).mean()), int(mask.sum())


	def _candidate_models(keys: List[str], allow_tree: bool = True):
	"""Yields (name, fit_fn) pairs. fit_fn(X, y) -> ScoreModel."""
	for c in CANDIDATE_C:
	def fit_fn(X, y, c=c):
	clf = LogisticRegression(max_iter=1000, C=c)
	clf.fit(X, y)
	return LinearModel({k: float(coef) for k, coef in zip(keys, clf.coef_[0])})
	yield f"logistic_regression(C={c})", fit_fn

	if not allow_tree:
	return

	for params in CANDIDATE_GBM_PARAMS:
	def fit_fn(X, y, params=params):
	clf = HistGradientBoostingClassifier(random_state=42, **params)
	clf.fit(X, y)
	return TreeModel(clf, keys)
	yield f"gradient_boosting({params})", fit_fn


	def fit_model(
	df: pd.DataFrame, lookahead: int, base_weights: Dict[str, float], allow_tree: bool = True
	) -> Tuple[ScoreModel, float, Dict]:
	"""Fits several candidate models (logistic regression at various
	regularization strengths, gradient boosting at a few conservative
	configs), but only adopts one if it actually beats the static
	`base_weights` on a final holdout that was never touched during
	fitting — otherwise falls back to `base_weights`, so this can never
	make accuracy worse than the static baseline.

	Layout (chronological, oldest to newest):
	[ inner-train \| inner-val \| final holdout ]
	inner-train/inner-val pick the best candidate model and score
	threshold; the final holdout is only ever scored once, to decide
	whether to keep the fitted model or the static one, and to report
	the accuracy shown to the user.

	Falls back to `LinearModel(base_weights)` and threshold 0.15 when there
	isn't enough history yet (new listings, short candle windows) or
	scikit-learn isn't installed. The returned diagnostics include
	`split_index`: callers should backtest only rows from `split_index`
	onward (the final holdout) to reproduce the same accuracy figure.

	`allow_tree=False` skips the gradient-boosting candidate entirely,
	leaving only logistic regression — used by the signals scanner
	(backend/scan/engine.py), which needs a JSON-serializable weights dict
	to cache in the database rather than a pickled sklearn object.
	"""
	fallback = (LinearModel(base_weights), 0.15, {"method": "static", "split_index": 0})
	if not _SKLEARN_AVAILABLE:
	return fallback

	vote_maps, y, keys = _build_training_rows(df, lookahead)
	n = len(vote_maps)
	if n < MIN_TRAIN_ROWS + MIN_TEST_ROWS or len(np.unique(y)) < 2:
	return fallback

	holdout_start = int(n * 0.7)
	inner_val_start = int(holdout_start * 0.7)
	if len(np.unique(y[:inner_val_start])) < 2 or (holdout_start - inner_val_start) < 15:
	return fallback

	X = np.array([[vm.get(k, 0.0) for k in keys] for vm in vote_maps])

	inner_train_vm, inner_val_vm = vote_maps[:inner_val_start], vote_maps[inner_val_start:holdout_start]
	inner_train_y, inner_val_y = y[:inner_val_start], y[inner_val_start:holdout_start]

	best_name, best_fit_fn, best_val_acc = None, None, -1.0
	for name, fit_fn in _candidate_models(keys, allow_tree):
	model = fit_fn(X[:inner_val_start], inner_train_y)
	train_scores = _score_rows(inner_train_vm, model)
	t = _best_threshold(train_scores, inner_train_y)
	val_acc, val_signals = _accuracy_on(inner_val_vm, inner_val_y, model, t)
	if val_signals >= MIN_VAL_SIGNALS and val_acc > best_val_acc:
	best_val_acc, best_name, best_fit_fn = val_acc, name, fit_fn

	if best_fit_fn is None:
	return fallback

	# Refit on the full pre-holdout region (inner-train + inner-val) with the
	# winning candidate's config, using all data available before the
	# untouched final holdout.
	fitted_model = best_fit_fn(X[:holdout_start], y[:holdout_start])
	fit_vm, fit_y = vote_maps[:holdout_start], y[:holdout_start]
	fitted_scores = _score_rows(fit_vm, fitted_model)
	fitted_threshold = _best_threshold(fitted_scores, fit_y)

	holdout_vm, holdout_y = vote_maps[holdout_start:], y[holdout_start:]
	fitted_acc, fitted_signals = _accuracy_on(holdout_vm, holdout_y, fitted_model, fitted_threshold)
	static_model = LinearModel(base_weights)
	static_acc, static_signals = _accuracy_on(holdout_vm, holdout_y, static_model, 0.15)

	diagnostics = {
	"split_index": holdout_start,
	"candidate": best_name,
	"fitted_holdout_accuracy": round(fitted_acc * 100, 1),
	"fitted_holdout_signals": fitted_signals,
	"static_holdout_accuracy": round(static_acc * 100, 1),
	"static_holdout_signals": static_signals,
	}

	if fitted_signals >= MIN_HOLDOUT_SIGNALS and fitted_acc > static_acc:
	diagnostics["method"] = "gradient_boosting" if isinstance(fitted_model, TreeModel) else "logistic_regression"
	return fitted_model, fitted_threshold, diagnostics

	diagnostics["method"] = "static"
	diagnostics["reason"] = "fitted model did not beat static baseline out-of-sample"
	return static_model, 0.15, diagnostics


	def run_backtest(
	df: pd.DataFrame,
	model,
	lookahead: int,
	threshold: float = 0.15,
	start_index: int = 0,
	) -> Dict:
	"""Backtests `model`/`threshold` from `start_index` onward.

	Pass `start_index=diagnostics["split_index"]` from `fit_model` to
	measure accuracy only on the held-out test portion (out-of-sample).
	Defaults to 0 (whole-history backtest) for static/fallback weights.
	`model` may be a plain weights dict or any ScoreModel (see scorer.py).
	"""
	closes = df["close"].to_numpy()
	n = len(df)
	correct = 0
	total = 0

	for i in range(start_index, n - lookahead):
	row = df.iloc[i]
	score, signal, _ = compute_signal(row, model, news_score=None, threshold=threshold)
	if signal == "NEUTRAL":
	continue

	current = closes[i]
	future = closes[i + lookahead]

	if signal == "BULLISH" and future > current:
	correct += 1
	elif signal == "BEARISH" and future < current:
	correct += 1
	total += 1

	accuracy = round(correct / total * 100, 1) if total > 0 else 0.0
	return {"accuracy": accuracy, "total_signals": total, "correct_signals": correct}