Skip to content

Beta Regime Screener

Screen NASDAQ stocks for market regime based on relative strength analysis.

Screener Functions

beta_regime

Beta regime screener for NASDAQ stocks.

Screens stocks for risk-on/risk-off regime based on Mike McGlone's methodology: - Relative strength ratio vs benchmark (e.g., SPY) - Ratio compared to its 200-day (or 40-week) moving average - Above MA = risk-on, Below MA = risk-off

BetaRegimeResult dataclass

Result from beta regime screening.

Attributes:

Name Type Description
ticker str

Stock symbol.

company_name str

Full company name or ticker if unavailable.

benchmark str

Benchmark ticker used for comparison.

regime Literal['risk-on', 'risk-off', 'insufficient-data']

Current regime status ("risk-on", "risk-off", or "insufficient-data").

relative_strength float

Asset price / benchmark price ratio.

ma_value float

Moving average of relative strength ratio.

pct_from_ma float

Percentage distance from moving average.

beta float

Rolling beta coefficient vs benchmark.

close_price float

Current asset closing price.

benchmark_price float

Current benchmark closing price.

interval str

Candle interval used ("1d", "1wk", "1mo").

ma_period int

Moving average period used for regime detection.

Source code in src/stockcharts/screener/beta_regime.py
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
@dataclass
class BetaRegimeResult:
    """Result from beta regime screening.

    Attributes:
        ticker: Stock symbol.
        company_name: Full company name or ticker if unavailable.
        benchmark: Benchmark ticker used for comparison.
        regime: Current regime status ("risk-on", "risk-off", or "insufficient-data").
        relative_strength: Asset price / benchmark price ratio.
        ma_value: Moving average of relative strength ratio.
        pct_from_ma: Percentage distance from moving average.
        beta: Rolling beta coefficient vs benchmark.
        close_price: Current asset closing price.
        benchmark_price: Current benchmark closing price.
        interval: Candle interval used ("1d", "1wk", "1mo").
        ma_period: Moving average period used for regime detection.
    """

    ticker: str
    company_name: str
    benchmark: str
    regime: Literal["risk-on", "risk-off", "insufficient-data"]
    relative_strength: float
    ma_value: float
    pct_from_ma: float
    beta: float
    close_price: float
    benchmark_price: float
    interval: str
    ma_period: int

screen_beta_regime(tickers=None, benchmark='SPY', interval='1d', ma_period=200, beta_window=60, regime_filter='all', min_price=None, max_price=None, min_volume=None, lookback=None, start=None, end=None, batch_size=50, verbose=True)

Screen stocks for beta regime status.

Parameters:

Name Type Description Default
tickers list[str] | None

List of ticker symbols (if None, uses all NASDAQ).

None
benchmark str

Benchmark ticker for comparison (default: "SPY"). Can be any valid yfinance ticker (SPY, QQQ, BTC-USD, etc.).

'SPY'
interval str

Candle interval - "1d", "1wk", "1mo" (default: "1d").

'1d'
ma_period int

Moving average period for regime detection. Default: 200 for daily, auto-adjusts to 40 for weekly if 200 is passed.

200
beta_window int

Rolling window for beta calculation (default: 60).

60
regime_filter Literal['risk-on', 'risk-off', 'all']

Filter results by regime - "risk-on", "risk-off", or "all".

'all'
min_price float | None

Minimum stock price filter.

None
max_price float | None

Maximum stock price filter.

None
min_volume float | None

Minimum average daily volume filter.

None
lookback str | None

Historical period (e.g., "1y", "2y", "5y"). Should be longer than ma_period to ensure sufficient data.

None
start str | None

Start date YYYY-MM-DD (overrides lookback if end also provided).

None
end str | None

End date YYYY-MM-DD.

None
batch_size int | None

Tickers per batch for parallel download (default: 50). Set to None for sequential mode.

50
verbose bool

Print progress messages (default: True).

True

Returns:

Type Description
list[BetaRegimeResult]

List of BetaRegimeResult objects.

Source code in src/stockcharts/screener/beta_regime.py
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
def screen_beta_regime(
    tickers: list[str] | None = None,
    benchmark: str = "SPY",
    interval: str = "1d",
    ma_period: int = 200,
    beta_window: int = 60,
    regime_filter: Literal["risk-on", "risk-off", "all"] = "all",
    min_price: float | None = None,
    max_price: float | None = None,
    min_volume: float | None = None,
    lookback: str | None = None,
    start: str | None = None,
    end: str | None = None,
    batch_size: int | None = 50,
    verbose: bool = True,
) -> list[BetaRegimeResult]:
    """Screen stocks for beta regime status.

    Args:
        tickers: List of ticker symbols (if None, uses all NASDAQ).
        benchmark: Benchmark ticker for comparison (default: "SPY").
            Can be any valid yfinance ticker (SPY, QQQ, BTC-USD, etc.).
        interval: Candle interval - "1d", "1wk", "1mo" (default: "1d").
        ma_period: Moving average period for regime detection.
            Default: 200 for daily, auto-adjusts to 40 for weekly if 200 is passed.
        beta_window: Rolling window for beta calculation (default: 60).
        regime_filter: Filter results by regime - "risk-on", "risk-off", or "all".
        min_price: Minimum stock price filter.
        max_price: Maximum stock price filter.
        min_volume: Minimum average daily volume filter.
        lookback: Historical period (e.g., "1y", "2y", "5y"). Should be longer
            than ma_period to ensure sufficient data.
        start: Start date YYYY-MM-DD (overrides lookback if end also provided).
        end: End date YYYY-MM-DD.
        batch_size: Tickers per batch for parallel download (default: 50).
            Set to None for sequential mode.
        verbose: Print progress messages (default: True).

    Returns:
        List of BetaRegimeResult objects.
    """
    # Auto-adjust MA period for weekly interval
    effective_ma_period = ma_period
    if interval == "1wk" and ma_period == 200:
        effective_ma_period = 40  # 40 weeks ≈ 200 days
        if verbose:
            print(f"Auto-adjusted MA period to {effective_ma_period} for weekly interval")

    # Auto-set lookback if not specified
    if lookback is None and start is None:
        # Need enough history for MA calculation + buffer
        if interval == "1d":
            lookback = "2y" if effective_ma_period <= 200 else "5y"
        elif interval == "1wk":
            lookback = "3y" if effective_ma_period <= 52 else "5y"
        else:
            lookback = "5y"

    # Get ticker list
    if tickers is None:
        if verbose:
            print("Fetching NASDAQ ticker list...")
        tickers = get_nasdaq_tickers()
        if verbose:
            print(f"Found {len(tickers)} tickers to screen")

    # Fetch benchmark data first
    if verbose:
        print(f"Fetching benchmark data ({benchmark})...")

    try:
        benchmark_df = fetch_ohlc(
            benchmark,
            interval=interval,
            lookback=lookback if not (start and end) else None,
            start=start,
            end=end,
        )
    except Exception as e:
        if verbose:
            print(f"Error fetching benchmark {benchmark}: {e}")
        return []

    if benchmark_df is None or benchmark_df.empty:
        if verbose:
            print(f"No data for benchmark {benchmark}")
        return []

    # Validate sufficient bars for MA calculation
    bars_available = len(benchmark_df)
    min_recommended = int(effective_ma_period * 1.5)  # 50% buffer for meaningful trend
    if bars_available < effective_ma_period:
        if verbose:
            print(
                f"⚠️  WARNING: Only {bars_available} bars available, but MA period is "
                f"{effective_ma_period}. No regime signals possible."
            )
            print("    Increase --period (e.g., 5y, 10y) or reduce --ma-period")
        return []
    elif bars_available < min_recommended:
        if verbose:
            print(
                f"⚠️  WARNING: Only {bars_available} bars for {effective_ma_period}-period MA. "
                f"Recommend {min_recommended}+ bars for reliable regime detection."
            )
            print("    Consider using --period 5y or --period 10y for better accuracy.")

    if verbose:
        print(f"Benchmark {benchmark}: {len(benchmark_df)} bars loaded")
        mode_msg = f"batch size {batch_size}" if batch_size else "sequential"
        print(f"Screening {len(tickers)} tickers ({mode_msg})...")
        print("-" * 70)

    # Screen tickers
    if batch_size is not None and batch_size > 0:
        results = _screen_batch_mode(
            tickers=tickers,
            benchmark=benchmark,
            benchmark_df=benchmark_df,
            interval=interval,
            lookback=lookback,
            start=start,
            end=end,
            effective_ma_period=effective_ma_period,
            beta_window=beta_window,
            regime_filter=regime_filter,
            min_price=min_price,
            max_price=max_price,
            min_volume=min_volume,
            batch_size=batch_size,
            verbose=verbose,
        )
    else:
        results = _screen_sequential_mode(
            tickers=tickers,
            benchmark=benchmark,
            benchmark_df=benchmark_df,
            interval=interval,
            lookback=lookback,
            start=start,
            end=end,
            effective_ma_period=effective_ma_period,
            beta_window=beta_window,
            regime_filter=regime_filter,
            min_price=min_price,
            max_price=max_price,
            min_volume=min_volume,
            verbose=verbose,
        )

    if verbose:
        print("-" * 70)
        print(f"Screening complete. Found {len(results)} stocks matching criteria.")

    return sorted(results, key=lambda x: x.ticker)

save_results_to_csv(results, filename='beta_regime_results.csv')

Save screening results to CSV file.

Parameters:

Name Type Description Default
results list[BetaRegimeResult]

List of BetaRegimeResult objects to save.

required
filename str

Output CSV file path.

'beta_regime_results.csv'

Returns:

Type Description
None

None. Prints status message and writes file to disk.

Source code in src/stockcharts/screener/beta_regime.py
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
def save_results_to_csv(
    results: list[BetaRegimeResult],
    filename: str = "beta_regime_results.csv",
) -> None:
    """Save screening results to CSV file.

    Args:
        results: List of BetaRegimeResult objects to save.
        filename: Output CSV file path.

    Returns:
        None. Prints status message and writes file to disk.
    """
    if not results:
        print("No results to save.")
        return

    df = pd.DataFrame(
        [
            {
                "Ticker": r.ticker,
                "Company_Name": r.company_name,
                "Benchmark": r.benchmark,
                "Regime": r.regime,
                "Relative_Strength": r.relative_strength,
                "MA_Value": r.ma_value,
                "Pct_From_MA": r.pct_from_ma,
                "Beta": r.beta,
                "Close_Price": r.close_price,
                "Benchmark_Price": r.benchmark_price,
                "Interval": r.interval,
                "MA_Period": r.ma_period,
            }
            for r in results
        ]
    )

    df.to_csv(filename, index=False)
    print(f"Results saved to {filename}")