tradingview-screener
Screen markets across 6 asset classes using TradingView data. API pre-filters + pandas computed signals. YAML-driven strategies.
Packaged view
This page reorganizes the original catalog entry around fit, installability, and workflow context first. The original raw source lives below.
Install command
npx @skill-hub/cli install openclaw-skills-tradingview-screener
Repository
Skill path: skills/hiehoo/tradingview-screener
Screen markets across 6 asset classes using TradingView data. API pre-filters + pandas computed signals. YAML-driven strategies.
Open repositoryBest for
Primary workflow: Analyze Data & AI.
Technical facets: Full Stack, Backend, Data / AI.
Target audience: everyone.
License: Unknown.
Original source
Catalog source: SkillHub Club.
Repository owner: openclaw.
This is still a mirrored public skill entry. Review the repository before installing into production workflows.
What it helps with
- Install tradingview-screener into Claude Code, Codex CLI, Gemini CLI, or OpenCode workflows
- Review https://github.com/openclaw/skills before adding tradingview-screener to shared team environments
- Use tradingview-screener for development workflows
Works across
Favorites: 0.
Sub-skills: 0.
Aggregator: No.
Original source / Raw SKILL.md
---
name: tradingview-screener
description: Screen markets across 6 asset classes using TradingView data. API pre-filters + pandas computed signals. YAML-driven strategies.
version: 1.1.0
---
# TradingView Screener
Screen stocks, crypto, forex, bonds, futures, and coins using TradingView's market data. Zero auth required.
## Setup (Run Once)
Before first use, run the install script to create a venv and install dependencies:
```bash
bash skills/tradingview-screener/install.sh
```
This creates `.venv/` inside the skill directory with all required packages.
## Execution
All scripts use the skill's own venv:
```bash
skills/tradingview-screener/.venv/bin/python3 skills/tradingview-screener/scripts/<script>.py [args]
```
**Windows:**
```bash
skills/tradingview-screener/.venv/Scripts/python.exe skills/tradingview-screener/scripts/<script>.py [args]
```
## Modes
| Mode | Description | Script |
|------|-------------|--------|
| **Screen** | One-time scan with filters, columns, sort | `screen.py` |
| **Signal** | YAML-driven signal detection (pre-filters + computed signals) | `signal-engine.py` |
## Quick Start
### Screen Mode
```bash
skills/tradingview-screener/.venv/bin/python3 skills/tradingview-screener/scripts/screen.py \
--asset-class stock --limit 20 \
--filters '[{"field":"MARKET_CAPITALIZATION","op":">","value":1000000000}]' \
--columns NAME,PRICE,CHANGE_PERCENT,VOLUME \
--sort-by VOLUME --sort-order desc
```
### Signal Mode
```bash
# List available signals
skills/tradingview-screener/.venv/bin/python3 skills/tradingview-screener/scripts/signal-engine.py --list
# Run a signal
skills/tradingview-screener/.venv/bin/python3 skills/tradingview-screener/scripts/signal-engine.py --signal golden-cross
```
## Asset Classes
| Class | Screener | Field Enum |
|-------|----------|------------|
| stock | StockScreener | StockField |
| crypto | CryptoScreener | CryptoField |
| forex | ForexScreener | ForexField |
| bond | BondScreener | BondField |
| futures | FuturesScreener | FuturesField |
| coin | CoinScreener | CoinField |
## Signal Types (Computed)
| Type | Description | Key Params |
|------|-------------|------------|
| crossover | Fast field crosses slow field | fast, slow, direction |
| threshold | Field crosses a value | field, op, value |
| expression | Pandas expression on DataFrame | expr |
| range | Field between min/max bounds | field, min, max |
## Filter Operators
`>`, `>=`, `<`, `<=`, `==`, `!=`, `between` (value: [min, max]), `isin` (value: [...])
## Common Stock Fields
`NAME`, `PRICE`, `CHANGE_PERCENT`, `VOLUME`, `MARKET_CAPITALIZATION`, `SECTOR`,
`SIMPLE_MOVING_AVERAGE_50`, `SIMPLE_MOVING_AVERAGE_200`, `RELATIVE_STRENGTH_INDEX_14`,
`MACD_LEVEL_12_26`, `AVERAGE_VOLUME_30_DAY`
Use `StockField.search("keyword")` in Python to discover more fields (13,000+ available).
## Pre-built Signals
| Signal | File | Description |
|--------|------|-------------|
| Golden Cross | `state/signals/golden-cross.yaml` | SMA50 above SMA200 (bullish) |
| Oversold Bounce | `state/signals/oversold-bounce.yaml` | RSI < 30 + price rising |
| Volume Breakout | `state/signals/volume-breakout.yaml` | Volume > 2x avg + momentum |
## Output Format
```markdown
**Stock Screener** | 15 results | Sorted by VOLUME desc
| NAME | PRICE | CHANGE_PERCENT | VOLUME |
|------|-------|----------------|--------|
| AAPL | 185.50 | 2.3 | 80000000 |
...
```
## Timeframes
`1`, `5`, `15`, `30`, `60`, `120`, `240`, `1D`, `1W`, `1M`
Pass `--timeframe 60` to apply hourly interval to technical indicators.
## References
- [API Guide](references/tvscreener-api-guide.md) - Screener types, filters, field discovery
- [Signals Guide](references/computed-signals-guide.md) - YAML schema, signal type configs
- [Strategy Templates](references/strategy-templates.md) - Pre-built screening strategies
- [Field Presets](references/field-presets.md) - Common field groups per asset class
---
## Referenced Files
> The following files are referenced in this skill and included for context.
### references/tvscreener-api-guide.md
```markdown
# TradingView Screener API Guide
## Screener Types
| Asset Class | Screener Class | Field Enum | Description |
|------------|---------------|------------|-------------|
| Stock | `stock.Screener()` | `StockField` | Equities and stocks |
| Crypto | `crypto.Screener()` | `CryptoField` | Cryptocurrencies |
| Forex | `forex.Screener()` | `ForexField` | Currency pairs |
| Bond | `bond.Screener()` | `BondField` | Bonds and fixed income |
| Futures | `futures.Screener()` | `FuturesField` | Futures contracts |
| Coin | `coin.Screener()` | `CoinField` | Coins and tokens |
## Filter Operators
| Operator | Symbol | Example Usage |
|----------|--------|---------------|
| Greater than | `>` | `filter(StockField.PRICE > 50)` |
| Greater or equal | `>=` | `filter(StockField.PRICE >= 100)` |
| Less than | `<` | `filter(StockField.VOLUME < 1000000)` |
| Less or equal | `<=` | `filter(StockField.CHANGE_PERCENT <= 5)` |
| Equal | `==` | `filter(StockField.SECTOR == "Technology")` |
| Not equal | `!=` | `filter(StockField.SECTOR != "Energy")` |
| Between | `between` | `filter(StockField.PRICE.between(10, 50))` |
| In list | `isin` | `filter(StockField.SECTOR.isin(["Technology", "Healthcare"]))` |
## Code Examples
### Basic Filter
```python
from tvscreener import stock, StockField
screener = stock.Screener()
screener.filter(StockField.PRICE > 100)
screener.filter(StockField.VOLUME > 1000000)
results = screener.get_scanner_data()
```
### Multiple Filters
```python
screener = stock.Screener()
screener.filter(StockField.PRICE.between(50, 200))
screener.filter(StockField.CHANGE_PERCENT > 2)
screener.filter(StockField.SECTOR.isin(["Technology", "Healthcare"]))
results = screener.get_scanner_data()
```
### Column Selection
By default, screeners return: NAME, PRICE, VOLUME, CHANGE_PERCENT
Add custom columns with `.select()`:
```python
screener = stock.Screener()
screener.filter(StockField.PRICE > 50)
screener.select(
StockField.NAME,
StockField.PRICE,
StockField.MARKET_CAPITALIZATION,
StockField.SIMPLE_MOVING_AVERAGE_50,
StockField.RELATIVE_STRENGTH_INDEX_14
)
results = screener.get_scanner_data()
```
## Timeframes
Supported intervals for technical indicators:
| Code | Description |
|------|-------------|
| "1" | 1 minute |
| "5" | 5 minutes |
| "15" | 15 minutes |
| "30" | 30 minutes |
| "60" | 1 hour |
| "120" | 2 hours |
| "240" | 4 hours |
| "1D" | Daily |
| "1W" | Weekly |
| "1M" | Monthly |
### Usage
```python
screener = stock.Screener()
screener.with_interval("1D")
screener.filter(StockField.RELATIVE_STRENGTH_INDEX_14 < 30)
results = screener.get_scanner_data()
```
## Field Discovery
### Search by Keyword
```python
# Find all fields containing "moving"
fields = StockField.search("moving")
for field in fields:
print(field.name, field.description)
```
### Technical Indicators
```python
# List all technical indicator fields
technicals = StockField.technicals()
```
### Fundamental Metrics
```python
# List all fundamental fields (stocks only)
fundamentals = StockField.fundamentals()
```
### Total Available Fields
```python
# StockField has 13,000+ fields
all_fields = StockField.all()
print(f"Total fields: {len(all_fields)}")
```
## Market Filtering
Filter by specific exchanges or markets:
```python
screener = stock.Screener()
screener.set_markets("america") # US markets
# or
screener.set_markets("NYSE", "NASDAQ") # Specific exchanges
results = screener.get_scanner_data()
```
Common market values:
- `"america"` - US markets
- `"NYSE"`, `"NASDAQ"`, `"AMEX"` - US exchanges
- `"LSE"` - London Stock Exchange
- `"TSX"` - Toronto Stock Exchange
## Error Handling Patterns
```python
from tvscreener import stock, StockField
import logging
logger = logging.getLogger(__name__)
def safe_screen(filters):
try:
screener = stock.Screener()
for field, op, value in filters:
if op == ">":
screener.filter(field > value)
elif op == "<":
screener.filter(field < value)
# ... other operators
results = screener.get_scanner_data()
return results
except ValueError as e:
logger.error(f"Invalid filter value: {e}")
return None
except Exception as e:
logger.error(f"Screener error: {e}")
return None
```
## Performance Tips
1. **Limit results**: Use `.limit(n)` to cap results
2. **Select specific columns**: Don't fetch all 13,000 fields
3. **Cache field lookups**: Store StockField.search() results
4. **Batch filters**: Combine related filters in one screener
5. **Use appropriate intervals**: Smaller intervals = more data
```
### references/computed-signals-guide.md
```markdown
# Computed Signals Guide
## Overview
Computed signals enable automated detection of trading opportunities based on technical indicators and price patterns. Signals are defined in YAML configuration files.
## YAML Schema
### Required Keys
- `name` - Signal identifier (string)
- `type` - Signal type: crossover, threshold, expression, range
- Type-specific parameters (see below)
### Optional Keys
- `description` - Human-readable explanation
- `timeframe` - Chart interval: "1D", "1W", "1M", etc.
- `min_volume` - Minimum daily volume filter
- `sector` - Sector filter (stocks only)
## Signal Types
### 1. Crossover Signals
Detects when one indicator crosses above/below another.
**Parameters:**
- `fast` - Fast-moving indicator field name
- `slow` - Slow-moving indicator field name
- `direction` - "up", "down", or "both"
**Example: Golden Cross**
```yaml
name: golden_cross
type: crossover
description: SMA50 crosses above SMA200 (bullish)
fast: SIMPLE_MOVING_AVERAGE_50
slow: SIMPLE_MOVING_AVERAGE_200
direction: up
timeframe: 1D
```
**Example: Death Cross**
```yaml
name: death_cross
type: crossover
description: SMA50 crosses below SMA200 (bearish)
fast: SIMPLE_MOVING_AVERAGE_50
slow: SIMPLE_MOVING_AVERAGE_200
direction: down
timeframe: 1D
```
### 2. Threshold Signals
Triggers when a field meets a threshold condition.
**Parameters:**
- `field` - Field name to evaluate
- `op` - Operator: ">", ">=", "<", "<=", "==", "!="
- `value` - Threshold value (numeric or string)
**Example: Oversold RSI**
```yaml
name: rsi_oversold
type: threshold
description: RSI below 30 indicates oversold
field: RELATIVE_STRENGTH_INDEX_14
op: "<"
value: 30
timeframe: 1D
```
**Example: High Volume**
```yaml
name: high_volume
type: threshold
description: Volume exceeds 10-day average by 50%
field: VOLUME
op: ">"
value: 1.5 * AVERAGE_VOLUME_10_DAY
```
**Example: Technology Sector**
```yaml
name: tech_sector
type: threshold
description: Technology sector stocks only
field: SECTOR
op: "=="
value: "Technology"
```
### 3. Expression Signals
Custom logic using pandas eval() with whitelist validation.
**Parameters:**
- `expr` - Expression string using column names and allowed operators
**Allowed in Expressions:**
- Column names (must exist in screener data)
- Operators: `+`, `-`, `*`, `/`, `>`, `<`, `>=`, `<=`, `==`, `!=`
- Logical: `and`, `or`, `not`
- Numeric literals: `1`, `2.5`, `100`
- Methods: `.mean()`, `.std()`
**NOT Allowed:**
- Function calls (except `.mean()`, `.std()`)
- Imports or external references
- String operations
- Assignment operations
**Example: Price Momentum**
```yaml
name: price_momentum
type: expression
description: Price above both SMAs with positive change
expr: (PRICE > SIMPLE_MOVING_AVERAGE_50) and (PRICE > SIMPLE_MOVING_AVERAGE_200) and (CHANGE_PERCENT > 0)
timeframe: 1D
```
**Example: Volume Surge**
```yaml
name: volume_surge
type: expression
description: Volume exceeds 30-day average by 2x
expr: VOLUME > (AVERAGE_VOLUME_30_DAY * 2)
```
### 4. Range Signals
Detects values within a numeric range.
**Parameters:**
- `field` - Field name to evaluate
- `min` - Minimum value (inclusive)
- `max` - Maximum value (inclusive)
**Example: Neutral RSI**
```yaml
name: rsi_neutral
type: range
description: RSI in neutral zone (40-60)
field: RELATIVE_STRENGTH_INDEX_14
min: 40
max: 60
timeframe: 1D
```
**Example: Mid-Cap Stocks**
```yaml
name: midcap_stocks
type: range
description: Market cap between 2B and 10B
field: MARKET_CAPITALIZATION
min: 2000000000
max: 10000000000
```
## Combining Multiple Signals
Create complex strategies by combining signal files:
```yaml
# signals/momentum-breakout.yaml
name: momentum_breakout
type: expression
description: Combined momentum and volume breakout
expr: (PRICE > SIMPLE_MOVING_AVERAGE_50) and (RELATIVE_STRENGTH_INDEX_14 > 50) and (VOLUME > AVERAGE_VOLUME_10_DAY * 1.5)
timeframe: 1D
min_volume: 1000000
```
## Expression Whitelist Validation
The skill validates expressions before evaluation:
✅ **Valid:**
```python
PRICE > SIMPLE_MOVING_AVERAGE_50
(PRICE > 100) and (VOLUME > 1000000)
RELATIVE_STRENGTH_INDEX_14 < 30
CHANGE_PERCENT > CHANGE_PERCENT.mean()
```
❌ **Invalid:**
```python
import os # No imports
eval("malicious code") # No eval/exec
PRICE.apply(lambda x: x * 2) # No lambda
open("/etc/passwd") # No file operations
```
## Common Patterns
### Bullish Momentum
```yaml
name: bullish_momentum
type: expression
expr: (PRICE > SIMPLE_MOVING_AVERAGE_50) and (SIMPLE_MOVING_AVERAGE_50 > SIMPLE_MOVING_AVERAGE_200) and (RELATIVE_STRENGTH_INDEX_14 > 50)
```
### Bearish Reversal
```yaml
name: bearish_reversal
type: expression
expr: (PRICE < SIMPLE_MOVING_AVERAGE_50) and (RELATIVE_STRENGTH_INDEX_14 > 70) and (CHANGE_PERCENT < -2)
```
### Volume Confirmation
```yaml
name: volume_confirmation
type: expression
expr: (CHANGE_PERCENT > 2) and (VOLUME > AVERAGE_VOLUME_30_DAY * 1.5)
```
### Breakout Setup
```yaml
name: breakout_setup
type: expression
expr: (PRICE > SIMPLE_MOVING_AVERAGE_50 * 1.02) and (VOLUME > AVERAGE_VOLUME_10_DAY * 2)
```
```
### references/strategy-templates.md
```markdown
# Strategy Templates
## Pre-Built Trading Strategies
Six battle-tested strategy templates for common trading approaches.
---
## 1. Momentum Strategy
**Asset Class:** Stock
**Description:** Identifies stocks with strong upward price momentum, rising above key moving averages with healthy volume.
**Key Filters:**
- Price above 50-day SMA
- Price above 200-day SMA
- Positive daily change
- Volume above 10-day average
- RSI above 50 (not overbought)
**CLI Command:**
```bash
python scripts/screen.py \
--asset stock \
--filters "PRICE > SIMPLE_MOVING_AVERAGE_50" \
"PRICE > SIMPLE_MOVING_AVERAGE_200" \
"CHANGE_PERCENT > 0" \
"VOLUME > AVERAGE_VOLUME_10_DAY" \
"RELATIVE_STRENGTH_INDEX_14 > 50" \
"RELATIVE_STRENGTH_INDEX_14 < 70" \
--columns NAME PRICE CHANGE_PERCENT VOLUME RELATIVE_STRENGTH_INDEX_14 \
--timeframe 1D \
--limit 50
```
**Best For:** Bull markets, trending stocks, swing trading
---
## 2. Value Strategy
**Asset Class:** Stock
**Description:** Finds undervalued stocks with strong fundamentals and oversold technical conditions.
**Key Filters:**
- RSI below 40 (oversold)
- Price below 50-day SMA (potential reversal)
- Positive market cap (established companies)
- Volume above average (liquidity)
**CLI Command:**
```bash
python scripts/screen.py \
--asset stock \
--filters "RELATIVE_STRENGTH_INDEX_14 < 40" \
"PRICE < SIMPLE_MOVING_AVERAGE_50" \
"MARKET_CAPITALIZATION > 1000000000" \
"VOLUME > AVERAGE_VOLUME_30_DAY" \
--columns NAME PRICE CHANGE_PERCENT MARKET_CAPITALIZATION RELATIVE_STRENGTH_INDEX_14 SECTOR \
--timeframe 1D \
--limit 50
```
**Best For:** Contrarian trading, long-term value investing, market corrections
---
## 3. Breakout Strategy
**Asset Class:** Stock
**Description:** Catches stocks breaking above resistance with volume confirmation.
**Key Filters:**
- Price near or above 50-day SMA
- Volume surge (2x 10-day average)
- Positive change
- RSI above 50 (momentum)
**CLI Command:**
```bash
python scripts/screen.py \
--asset stock \
--filters "PRICE > SIMPLE_MOVING_AVERAGE_50 * 0.98" \
"VOLUME > AVERAGE_VOLUME_10_DAY * 2" \
"CHANGE_PERCENT > 2" \
"RELATIVE_STRENGTH_INDEX_14 > 50" \
--columns NAME PRICE CHANGE_PERCENT VOLUME SIMPLE_MOVING_AVERAGE_50 \
--timeframe 1D \
--limit 30
```
**Best For:** Day trading, momentum trading, volatility plays
---
## 4. Mean Reversion Strategy
**Asset Class:** Stock
**Description:** Identifies oversold stocks likely to bounce back to their moving averages.
**Key Filters:**
- RSI below 30 (oversold)
- Price significantly below 50-day SMA
- Established company (market cap filter)
- Normal volume
**CLI Command:**
```bash
python scripts/screen.py \
--asset stock \
--filters "RELATIVE_STRENGTH_INDEX_14 < 30" \
"PRICE < SIMPLE_MOVING_AVERAGE_50 * 0.95" \
"MARKET_CAPITALIZATION > 500000000" \
"VOLUME > AVERAGE_VOLUME_30_DAY * 0.5" \
--columns NAME PRICE CHANGE_PERCENT RELATIVE_STRENGTH_INDEX_14 SIMPLE_MOVING_AVERAGE_50 \
--timeframe 1D \
--limit 40
```
**Best For:** Short-term trading, range-bound markets, oversold bounces
---
## 5. Crypto Momentum Strategy
**Asset Class:** Crypto
**Description:** High-momentum cryptocurrencies with strong volume and positive trends.
**Key Filters:**
- Positive daily change
- Volume above average
- RSI above 50 (bullish momentum)
- Price above 50-period SMA
**CLI Command:**
```bash
python scripts/screen.py \
--asset crypto \
--filters "CHANGE_PERCENT > 3" \
"VOLUME > AVERAGE_VOLUME_10_DAY" \
"RELATIVE_STRENGTH_INDEX_14 > 50" \
"RELATIVE_STRENGTH_INDEX_14 < 75" \
"PRICE > SIMPLE_MOVING_AVERAGE_50" \
--columns NAME PRICE CHANGE_PERCENT VOLUME RELATIVE_STRENGTH_INDEX_14 \
--timeframe 1D \
--limit 30
```
**Best For:** Crypto trading, high volatility, trend following
---
## 6. Forex Trend Strategy
**Asset Class:** Forex
**Description:** Currency pairs with established trends and momentum confirmation.
**Key Filters:**
- 50-day SMA above 200-day SMA (golden cross)
- Price above 50-day SMA
- RSI between 45-65 (trending, not extreme)
**CLI Command:**
```bash
python scripts/screen.py \
--asset forex \
--filters "SIMPLE_MOVING_AVERAGE_50 > SIMPLE_MOVING_AVERAGE_200" \
"PRICE > SIMPLE_MOVING_AVERAGE_50" \
"RELATIVE_STRENGTH_INDEX_14 > 45" \
"RELATIVE_STRENGTH_INDEX_14 < 65" \
--columns NAME PRICE CHANGE_PERCENT SIMPLE_MOVING_AVERAGE_50 SIMPLE_MOVING_AVERAGE_200 \
--timeframe 1D \
--limit 20
```
**Best For:** Forex trading, trend following, position trading
---
## Usage Tips
### Combining Strategies
Mix filters from different strategies:
```bash
# Momentum + Value hybrid
python scripts/screen.py \
--asset stock \
--filters "PRICE > SIMPLE_MOVING_AVERAGE_50" \
"MARKET_CAPITALIZATION > 2000000000" \
"RELATIVE_STRENGTH_INDEX_14 < 60" \
--limit 50
```
### Adjusting Timeframes
Change timeframes for different trading styles:
- **Day trading:** `--timeframe 5` or `--timeframe 15`
- **Swing trading:** `--timeframe 1D`
- **Position trading:** `--timeframe 1W`
### Market Filtering
Add market filters for geographic focus:
```bash
python scripts/screen.py \
--asset stock \
--filters "PRICE > SIMPLE_MOVING_AVERAGE_50" \
--markets america \
--limit 50
```
### Sector Filtering
Focus on specific sectors:
```bash
python scripts/screen.py \
--asset stock \
--filters "PRICE > SIMPLE_MOVING_AVERAGE_50" \
"SECTOR == Technology" \
--limit 50
```
## Risk Warnings
⚠️ **Important:**
- These strategies are templates, not trading advice
- Always backtest before live trading
- Use proper risk management
- Past performance ≠ future results
- Consider transaction costs and slippage
```
### references/field-presets.md
```markdown
# Field Presets
## Quick Reference for Common TradingView Fields
All field names are **VERIFIED** from actual tvscreener introspection. Use full field names - short aliases do NOT exist.
---
## Stock Fields
### Price & Volume
```python
StockField.NAME # Ticker symbol
StockField.PRICE # Current price
StockField.CHANGE_PERCENT # Daily % change
StockField.VOLUME # Current volume
StockField.AVERAGE_VOLUME_10_DAY # 10-day average volume
StockField.AVERAGE_VOLUME_30_DAY # 30-day average volume
```
### Valuation & Fundamentals
```python
StockField.MARKET_CAPITALIZATION # Market cap in USD
StockField.SECTOR # Business sector
```
### Technical Indicators - Moving Averages
```python
StockField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
StockField.SIMPLE_MOVING_AVERAGE_200 # 200-period SMA
```
### Technical Indicators - Momentum
```python
StockField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
StockField.MACD_LEVEL_12_26 # MACD line (12,26,9)
```
---
## Crypto Fields
### Price & Volume
```python
CryptoField.NAME # Symbol (BTC, ETH, etc.)
CryptoField.PRICE # Current price
CryptoField.CHANGE_PERCENT # Daily % change
CryptoField.VOLUME # 24h volume
CryptoField.AVERAGE_VOLUME_10_DAY # 10-day average volume
```
### Technical Indicators
```python
CryptoField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
CryptoField.SIMPLE_MOVING_AVERAGE_200 # 200-period SMA
CryptoField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
CryptoField.MACD_LEVEL_12_26 # MACD line
```
---
## Forex Fields
### Price & Change
```python
ForexField.NAME # Currency pair (EUR/USD, etc.)
ForexField.PRICE # Current rate
ForexField.CHANGE_PERCENT # Daily % change
```
### Technical Indicators
```python
ForexField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
ForexField.SIMPLE_MOVING_AVERAGE_200 # 200-period SMA
ForexField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
ForexField.MACD_LEVEL_12_26 # MACD line
```
---
## Bond Fields
### Price & Yield
```python
BondField.NAME # Bond identifier
BondField.PRICE # Current price
BondField.CHANGE_PERCENT # Daily % change
```
### Technical Indicators
```python
BondField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
BondField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
```
---
## Futures Fields
### Price & Volume
```python
FuturesField.NAME # Contract name
FuturesField.PRICE # Current price
FuturesField.CHANGE_PERCENT # Daily % change
FuturesField.VOLUME # Volume
```
### Technical Indicators
```python
FuturesField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
FuturesField.SIMPLE_MOVING_AVERAGE_200 # 200-period SMA
FuturesField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
```
---
## Coin Fields
### Price & Volume
```python
CoinField.NAME # Coin symbol
CoinField.PRICE # Current price
CoinField.CHANGE_PERCENT # Daily % change
CoinField.VOLUME # 24h volume
```
### Technical Indicators
```python
CoinField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
CoinField.RELATIVE_STRENGTH_INDEX_14 # 14-period RSI
```
---
## Common Technical Indicators (Stocks)
### Moving Averages
```python
StockField.SIMPLE_MOVING_AVERAGE_50 # 50-period SMA
StockField.SIMPLE_MOVING_AVERAGE_200 # 200-period SMA
# Note: EMA fields exist but naming pattern varies
# Use StockField.search("exponential") to discover
```
### Momentum Oscillators
```python
StockField.RELATIVE_STRENGTH_INDEX_14 # RSI (14-period)
StockField.MACD_LEVEL_12_26 # MACD line (12,26,9)
# Note: MACD signal and histogram have different field names
# Use StockField.search("macd") to discover
```
### Bollinger Bands
```python
# Use StockField.search("bollinger") to find:
# - Upper band
# - Middle band (SMA)
# - Lower band
```
### Average Directional Index (ADX)
```python
# Use StockField.search("directional") to find ADX fields
```
---
## Field Discovery Commands
### Search by Keyword
```python
from tvscreener import StockField
# Find all moving average fields
ma_fields = StockField.search("moving")
for field in ma_fields:
print(f"{field.name}: {field.description}")
```
### List Technical Indicators
```python
# All technical indicator fields
tech_fields = StockField.technicals()
print(f"Found {len(tech_fields)} technical fields")
```
### List Fundamental Metrics (Stocks Only)
```python
# All fundamental fields
fund_fields = StockField.fundamentals()
print(f"Found {len(fund_fields)} fundamental fields")
```
### Search Specific Indicators
```python
# RSI variations
rsi_fields = StockField.search("rsi")
# MACD variations
macd_fields = StockField.search("macd")
# Bollinger Bands
bb_fields = StockField.search("bollinger")
# Stochastic
stoch_fields = StockField.search("stochastic")
# Volume indicators
vol_fields = StockField.search("volume")
```
---
## Usage Examples
### Basic Screen with Price & Volume
```python
from tvscreener import stock, StockField
screener = stock.Screener()
screener.filter(StockField.PRICE > 50)
screener.filter(StockField.VOLUME > 1000000)
screener.select(
StockField.NAME,
StockField.PRICE,
StockField.VOLUME,
StockField.CHANGE_PERCENT
)
results = screener.get_scanner_data()
```
### Technical Analysis Screen
```python
screener = stock.Screener()
screener.with_interval("1D")
screener.filter(StockField.PRICE > StockField.SIMPLE_MOVING_AVERAGE_50)
screener.filter(StockField.RELATIVE_STRENGTH_INDEX_14 > 50)
screener.select(
StockField.NAME,
StockField.PRICE,
StockField.SIMPLE_MOVING_AVERAGE_50,
StockField.RELATIVE_STRENGTH_INDEX_14,
StockField.MACD_LEVEL_12_26
)
results = screener.get_scanner_data()
```
### Fundamental + Technical Screen
```python
screener = stock.Screener()
screener.filter(StockField.MARKET_CAPITALIZATION > 1000000000)
screener.filter(StockField.SECTOR == "Technology")
screener.filter(StockField.RELATIVE_STRENGTH_INDEX_14 < 70)
screener.select(
StockField.NAME,
StockField.PRICE,
StockField.MARKET_CAPITALIZATION,
StockField.SECTOR,
StockField.RELATIVE_STRENGTH_INDEX_14
)
results = screener.get_scanner_data()
```
---
## Important Notes
⚠️ **Field Name Accuracy:**
- Always use FULL field names (e.g., `SIMPLE_MOVING_AVERAGE_50`)
- Short aliases like `SMA50`, `RSI14` do NOT exist
- Use `StockField.search("keyword")` when unsure
⚠️ **Asset Class Differences:**
- Not all fields exist across all asset classes
- Fundamental fields (market cap, sector) only for stocks
- Always test field availability with `.search()`
⚠️ **13,000+ Fields Available:**
- This guide covers the most common fields
- Use discovery commands to explore all available fields
- Custom indicators and exotic metrics require field search
```
### state/signals/golden-cross.yaml
```yaml
# Golden Cross Signal
# Detects stocks where 50-day MA crosses above 200-day MA (bullish momentum)
name: golden-cross
description: "SMA50 crosses above SMA200 - bullish momentum with strong volume"
asset_class: stock
timeframe: "1D"
# Pre-filter: large-cap stocks with decent volume
filters:
- field: MARKET_CAPITALIZATION
op: ">"
value: 1000000000
- field: VOLUME
op: ">"
value: 500000
columns:
- NAME
- PRICE
- CHANGE_PERCENT
- SIMPLE_MOVING_AVERAGE_50
- SIMPLE_MOVING_AVERAGE_200
- VOLUME
# Signal: detect upward MA crossover
computed:
- type: crossover
fast: SIMPLE_MOVING_AVERAGE_50
slow: SIMPLE_MOVING_AVERAGE_200
direction: up
sort_by: VOLUME
sort_order: desc
limit: 20
```
### state/signals/oversold-bounce.yaml
```yaml
# Oversold Bounce Signal
# Detects oversold stocks (RSI < 30) showing early price recovery
name: oversold-bounce
description: "RSI below 30 with positive price change - potential bounce"
asset_class: stock
timeframe: "1D"
# Pre-filter: mid-cap+ stocks in oversold territory
filters:
- field: MARKET_CAPITALIZATION
op: ">"
value: 500000000
- field: RELATIVE_STRENGTH_INDEX_14
op: "<"
value: 35
columns:
- NAME
- PRICE
- CHANGE_PERCENT
- RELATIVE_STRENGTH_INDEX_14
- VOLUME
# Signals: deeply oversold + positive price change
computed:
- type: threshold
field: RELATIVE_STRENGTH_INDEX_14
op: "<"
value: 30
- type: expression
expr: "CHANGE_PERCENT > 0"
sort_by: RELATIVE_STRENGTH_INDEX_14
sort_order: asc
limit: 20
```
### state/signals/volume-breakout.yaml
```yaml
# Volume Breakout Signal
# Detects stocks with unusual volume surge (2x+ average) and strong price momentum
name: volume-breakout
description: "Volume > 2x 30-day average with strong price momentum"
asset_class: stock
timeframe: "1D"
# Pre-filter: large-cap stocks with momentum
filters:
- field: MARKET_CAPITALIZATION
op: ">"
value: 1000000000
- field: CHANGE_PERCENT
op: ">"
value: 2
columns:
- NAME
- PRICE
- CHANGE_PERCENT
- VOLUME
- AVERAGE_VOLUME_30_DAY
- RELATIVE_STRENGTH_INDEX_14
# Signals: volume spike + strong price action
computed:
- type: expression
expr: "VOLUME > AVERAGE_VOLUME_30_DAY * 2"
- type: threshold
field: CHANGE_PERCENT
op: ">"
value: 3
sort_by: CHANGE_PERCENT
sort_order: desc
limit: 20
```
---
## Skill Companion Files
> Additional files collected from the skill directory layout.
### _meta.json
```json
{
"owner": "hiehoo",
"slug": "tradingview-screener",
"displayName": "TradingView Screener",
"latest": {
"version": "1.1.0",
"publishedAt": 1770630705911,
"commit": "https://github.com/openclaw/skills/commit/4f21027bc9d1003898ef1c095fdc738a830dac75"
},
"history": []
}
```
### assets/signal-template.yaml
```yaml
# Signal Template - Copy and customize
# Save to state/signals/{your-signal-name}.yaml
name: my-signal # Required: unique signal name
description: "" # Required: what this signal detects
asset_class: stock # Required: stock|crypto|forex|bond|futures|coin
timeframe: "1D" # Optional: 1|5|15|30|60|120|240|1D|1W|1M
# API pre-filters (server-side, fast)
filters:
- field: MARKET_CAPITALIZATION
op: ">"
value: 1000000000
# - field: VOLUME
# op: ">="
# value: 500000
# Operators: > >= < <= == != between isin
# For between: value: [min, max]
# For isin: value: [val1, val2, ...]
# Columns to fetch (use full enum names)
columns:
- NAME
- PRICE
- CHANGE_PERCENT
- VOLUME
# More: SIMPLE_MOVING_AVERAGE_50, SIMPLE_MOVING_AVERAGE_200,
# RELATIVE_STRENGTH_INDEX_14, MACD_LEVEL_12_26,
# AVERAGE_VOLUME_30_DAY, MARKET_CAPITALIZATION, SECTOR
# Post-fetch computed signals (pandas, client-side)
computed:
# Crossover: detect MA/indicator crossovers
- type: crossover
fast: SIMPLE_MOVING_AVERAGE_50
slow: SIMPLE_MOVING_AVERAGE_200
direction: up # up|down|both
# Threshold: field vs value comparison
# - type: threshold
# field: RELATIVE_STRENGTH_INDEX_14
# op: "<"
# value: 30
# Expression: pandas eval expression (use enum-style field names)
# - type: expression
# expr: "VOLUME > AVERAGE_VOLUME_30_DAY * 2"
# Range: field between bounds
# - type: range
# field: RELATIVE_STRENGTH_INDEX_14
# min: 40
# max: 60
# Output control
sort_by: VOLUME
sort_order: desc # asc|desc
limit: 20 # Max results
```
### scripts/requirements.txt
```text
tvscreener>=0.2.0
pandas>=2.0.0
pyyaml>=6.0
pytest>=7.0.0
```
### scripts/screen.py
```python
"""TradingView market screener CLI. Fetches filtered data and outputs markdown."""
import argparse
import json
import sys
import os
sys.path.insert(0, os.path.dirname(__file__))
from screener_constants import SCREENER_MAP, OP_MAP, resolve_field, find_df_column
def apply_filters(screener, field_enum, filters_json, timeframe=None):
"""Apply JSON filter specs to a screener via .where() chains."""
if not filters_json:
return screener
try:
filters = json.loads(filters_json) if isinstance(filters_json, str) else filters_json
except json.JSONDecodeError as e:
print(f"Error: Invalid filter JSON: {e}", file=sys.stderr)
sys.exit(1)
for f in filters:
field = resolve_field(field_enum, f["field"])
op = f.get("op", ">")
value = f["value"]
if op not in OP_MAP:
print(f"Error: Unknown operator '{op}'", file=sys.stderr)
sys.exit(1)
if timeframe and timeframe not in ("1D", "1d") and hasattr(field, "with_interval"):
try:
field = field.with_interval(timeframe)
except Exception:
pass # Not all fields support intervals
screener.where(OP_MAP[op](field, value))
return screener
def select_columns(screener, field_enum, columns_str, timeframe=None):
"""Select columns on a screener from comma-separated field names."""
if not columns_str:
return screener
names = [c.strip() for c in columns_str.split(",") if c.strip()]
fields = []
for name in names:
field = resolve_field(field_enum, name)
if timeframe and timeframe not in ("1D", "1d") and hasattr(field, "with_interval"):
try:
field = field.with_interval(timeframe)
except Exception:
pass
fields.append(field)
if fields:
screener.select(*fields)
return screener
def format_markdown(df, asset_class, sort_by=None, sort_order="desc", limit=50):
"""Format DataFrame as markdown table with summary line."""
if df.empty:
return f"**{asset_class.title()} Screener** | 0 results\n\nNo matches found."
if sort_by:
col = find_df_column(df, sort_by)
if col:
ascending = sort_order.lower() == "asc"
df = df.sort_values(by=col, ascending=ascending)
total = len(df)
df = df.head(limit)
try:
table = df.to_markdown(index=False)
except ImportError:
# Fallback if tabulate not installed
table = df.to_string(index=False)
sort_info = f"Sorted by {sort_by} {sort_order}" if sort_by else "Default order"
summary = f"**{asset_class.title()} Screener** | {len(df)} of {total} results | {sort_info}"
return f"{summary}\n\n{table}"
def main():
parser = argparse.ArgumentParser(description="TradingView Market Screener")
parser.add_argument("--asset-class", default="stock",
choices=list(SCREENER_MAP.keys()),
help="Asset class to screen")
parser.add_argument("--filters", default=None,
help='JSON filter array, e.g. \'[{"field":"PRICE","op":">","value":50}]\'')
parser.add_argument("--columns", default=None,
help="Comma-separated column names")
parser.add_argument("--sort-by", default=None,
help="Column name to sort by")
parser.add_argument("--sort-order", default="desc", choices=["asc", "desc"])
parser.add_argument("--limit", type=int, default=50,
help="Max results to display")
parser.add_argument("--timeframe", default=None,
help="Interval for technical fields (1,5,15,30,60,120,240,1D,1W,1M)")
parser.add_argument("--market", default=None,
help="Market filter (e.g. america, europe)")
args = parser.parse_args()
try:
screener_cls, field_enum = SCREENER_MAP[args.asset_class]
screener = screener_cls()
if args.market:
screener.set_markets(args.market)
apply_filters(screener, field_enum, args.filters, args.timeframe)
select_columns(screener, field_enum, args.columns, args.timeframe)
df = screener.get()
output = format_markdown(df, args.asset_class, args.sort_by, args.sort_order, args.limit)
print(output)
except ValueError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Error: Screener failed: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/screener_constants.py
```python
"""Shared constants and helpers for TradingView screener scripts."""
import tvscreener as tvs
# Maps asset class string to (ScreenerClass, FieldEnum) tuple
SCREENER_MAP = {
"stock": (tvs.StockScreener, tvs.StockField),
"crypto": (tvs.CryptoScreener, tvs.CryptoField),
"forex": (tvs.ForexScreener, tvs.ForexField),
"bond": (tvs.BondScreener, tvs.BondField),
"futures": (tvs.FuturesScreener, tvs.FuturesField),
"coin": (tvs.CoinScreener, tvs.CoinField),
}
# Maps operator string to lambda applying the operation on a Field
OP_MAP = {
">": lambda f, v: f > v,
">=": lambda f, v: f >= v,
"<": lambda f, v: f < v,
"<=": lambda f, v: f <= v,
"==": lambda f, v: f == v,
"!=": lambda f, v: f != v,
"between": lambda f, v: f.between(v[0], v[1]),
"isin": lambda f, v: f.isin(v),
}
def resolve_field(field_enum, field_name):
"""Resolve a field name string to a Field enum instance.
Args:
field_enum: The Field enum class (e.g., StockField)
field_name: String name of the field (case-insensitive)
Returns:
Field enum instance
Raises:
ValueError: If field name not found in enum
"""
name_upper = field_name.strip().upper()
try:
return getattr(field_enum, name_upper)
except AttributeError:
# Try search as fallback
results = field_enum.search(field_name)
if results:
return results[0]
raise ValueError(
f"Field '{field_name}' not found in {field_enum.__name__}. "
f"Use {field_enum.__name__}.search('{field_name}') to discover fields."
)
def _normalize(s):
"""Normalize a string for fuzzy column matching.
Strips underscores, parens, percent signs, extra spaces, lowercases.
E.g., 'SIMPLE_MOVING_AVERAGE_50' → 'simple moving average 50'
'Simple Moving Average (50)' → 'simple moving average 50'
"""
import re
s = s.lower().replace("_", " ").replace("(", " ").replace(")", " ").replace("%", " ")
return re.sub(r"\s+", " ", s).strip()
def find_df_column(df, name):
"""Find a DataFrame column by name (case-insensitive, normalized match).
tvscreener returns display names (e.g., 'Volume') not enum names ('VOLUME').
This helper matches user-provided names against actual DataFrame columns.
"""
if name in df.columns:
return name
# Case-insensitive exact match
lower_map = {c.lower(): c for c in df.columns}
if name.lower() in lower_map:
return lower_map[name.lower()]
# Normalized match: strip underscores, parens, etc.
name_norm = _normalize(name)
for col in df.columns:
if _normalize(col) == name_norm:
return col
# Partial containment match
for col in df.columns:
col_norm = _normalize(col)
if name_norm in col_norm or col_norm in name_norm:
return col
return None
```
### scripts/signal_engine.py
```python
"""YAML-driven signal engine. Loads signal configs, runs screener, applies computed signals."""
import argparse
import json
import sys
import os
from pathlib import Path
import yaml
sys.path.insert(0, os.path.dirname(__file__))
from screener_constants import SCREENER_MAP, OP_MAP, resolve_field, find_df_column
from signal_types import apply_computed_signals
# Default signals directory relative to skill root
SKILL_ROOT = Path(__file__).resolve().parent.parent
DEFAULT_SIGNALS_DIR = SKILL_ROOT / "state" / "signals"
def load_signal(signal_path):
"""Load and validate a YAML signal config file."""
path = Path(signal_path)
if not path.exists():
# Try as name in default signals dir
path = DEFAULT_SIGNALS_DIR / f"{signal_path}.yaml"
if not path.exists():
raise FileNotFoundError(f"Signal config not found: {signal_path}")
with open(path) as f:
config = yaml.safe_load(f)
if not config or "name" not in config or "asset_class" not in config:
raise ValueError(f"Invalid signal config: missing 'name' or 'asset_class' in {path}")
return config
def list_signals(signals_dir=None):
"""List available signal configs as markdown table."""
sig_dir = Path(signals_dir) if signals_dir else DEFAULT_SIGNALS_DIR
if not sig_dir.exists():
print("No signals directory found.")
return
yamls = sorted(sig_dir.glob("*.yaml"))
if not yamls:
print("No signal configs found.")
return
print("| Signal | Description | Asset | Types |")
print("|--------|-------------|-------|-------|")
for yf in yamls:
try:
with open(yf) as f:
cfg = yaml.safe_load(f)
name = cfg.get("name", yf.stem)
desc = cfg.get("description", "")[:60]
asset = cfg.get("asset_class", "?")
types = ", ".join(c.get("type", "?") for c in cfg.get("computed", []))
print(f"| {name} | {desc} | {asset} | {types} |")
except Exception as e:
print(f"| {yf.stem} | Error: {e} | ? | ? |")
def build_and_run(config):
"""Build screener from config, fetch data, apply computed signals, output markdown."""
asset_class = config["asset_class"]
if asset_class not in SCREENER_MAP:
raise ValueError(f"Unknown asset class: {asset_class}. Options: {list(SCREENER_MAP.keys())}")
screener_cls, field_enum = SCREENER_MAP[asset_class]
screener = screener_cls()
# Only apply with_interval for non-daily timeframes (daily is default)
timeframe = config.get("timeframe")
use_interval = timeframe and timeframe not in ("1D", "1d")
# Apply API pre-filters
for f in config.get("filters", []):
field = resolve_field(field_enum, f["field"])
op = f.get("op", ">")
value = f["value"]
if op not in OP_MAP:
raise ValueError(f"Unknown operator: {op}")
if use_interval and hasattr(field, "with_interval"):
try:
field = field.with_interval(timeframe)
except Exception:
pass
screener.where(OP_MAP[op](field, value))
# Select columns
columns = config.get("columns", [])
if columns:
fields = []
for name in columns:
fld = resolve_field(field_enum, name)
if use_interval and hasattr(fld, "with_interval"):
try:
fld = fld.with_interval(timeframe)
except Exception:
pass
fields.append(fld)
screener.select(*fields)
# Fetch data
df = screener.get()
total_rows = len(df)
# Apply computed signals (post-filter)
computed = config.get("computed", [])
df = apply_computed_signals(df, computed)
# Sort
sort_by = config.get("sort_by")
if sort_by:
col = find_df_column(df, sort_by)
if col:
ascending = config.get("sort_order", "desc").lower() == "asc"
df = df.sort_values(by=col, ascending=ascending)
# Limit
limit = config.get("limit", 50)
df = df.head(limit)
# Format output
sig_name = config.get("name", "Signal")
sig_desc = config.get("description", "")
header = f"**{sig_name}** — {sig_desc}" if sig_desc else f"**{sig_name}**"
types_used = ", ".join(c.get("type", "?") for c in computed) if computed else "none"
summary = f"{header}\n{len(df)} matches from {total_rows} rows | Signals: {types_used}"
try:
table = df.to_markdown(index=False)
except ImportError:
table = df.to_string(index=False)
print(f"{summary}\n\n{table}")
def main():
parser = argparse.ArgumentParser(description="TradingView Signal Engine")
parser.add_argument("--signal", default=None,
help="Signal name (from state/signals/) or path to YAML file")
parser.add_argument("--list", action="store_true",
help="List available signal configs")
parser.add_argument("--signals-dir", default=None,
help="Override signals directory path")
args = parser.parse_args()
if args.list:
list_signals(args.signals_dir)
return
if not args.signal:
parser.print_help()
sys.exit(1)
try:
config = load_signal(args.signal)
build_and_run(config)
except (FileNotFoundError, ValueError) as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Error: Signal engine failed: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
```
### scripts/signal_types.py
```python
"""Signal type implementations for computed post-filter signals.
Each function takes a DataFrame + config dict, returns filtered DataFrame.
4 signal types: crossover, threshold, expression, range.
"""
import re
import sys
import os
import pandas as pd
sys.path.insert(0, os.path.dirname(__file__))
from screener_constants import find_df_column
# Whitelist for expression validation: column ops, basic math, comparisons
EXPR_ALLOWED_PATTERN = re.compile(
r"^[A-Za-z0-9_\s\.\+\-\*\/\>\<\=\!\(\)\,\%]+$"
)
EXPR_BLOCKED_KEYWORDS = {"import", "exec", "eval", "__", "open", "os", "sys", "lambda"}
def validate_expression(expr):
"""Validate expression string against whitelist rules.
Allows: column names, basic operators (+,-,*,/,>,<,>=,<=,==,!=,and,or,not),
numeric literals, .mean(), .std(), .sum(), .min(), .max()
"""
if not EXPR_ALLOWED_PATTERN.match(expr):
raise ValueError(f"Expression contains disallowed characters: {expr}")
words = set(re.findall(r"[a-zA-Z_]\w*", expr.lower()))
blocked = words & EXPR_BLOCKED_KEYWORDS
if blocked:
raise ValueError(f"Expression contains blocked keywords: {blocked}")
def apply_crossover(df, config):
"""Detect when fast field is above/below slow field.
Config keys: fast (column), slow (column), direction (up|down|both)
"""
fast = find_df_column(df, config["fast"])
slow = find_df_column(df, config["slow"])
direction = config.get("direction", "up")
if not fast or not slow:
print(f"Warning: Crossover columns not found in DataFrame. "
f"Available: {list(df.columns)}", file=sys.stderr)
return df
if direction == "up":
mask = df[fast] > df[slow]
elif direction == "down":
mask = df[fast] < df[slow]
else: # both
mask = df[fast] != df[slow]
return df[mask].copy()
def apply_threshold(df, config):
"""Filter rows where field crosses a threshold value.
Config keys: field (column), op (comparison), value (number)
"""
col = find_df_column(df, config["field"])
op = config.get("op", ">")
value = config["value"]
if not col:
print(f"Warning: Threshold field '{config['field']}' not in DataFrame. "
f"Available: {list(df.columns)}", file=sys.stderr)
return df
ops = {
">": lambda: df[col] > value,
">=": lambda: df[col] >= value,
"<": lambda: df[col] < value,
"<=": lambda: df[col] <= value,
"==": lambda: df[col] == value,
"!=": lambda: df[col] != value,
}
if op not in ops:
raise ValueError(f"Unknown threshold operator: {op}")
return df[ops[op]()].copy()
def apply_expression(df, config):
"""Evaluate a pandas expression on the DataFrame.
Config keys: expr (pandas eval string)
Uses df.eval() which is sandboxed to DataFrame operations.
Resolves enum-style field names to actual DataFrame column names.
"""
expr = config["expr"]
validate_expression(expr)
# Resolve field names in expression to actual DataFrame column names
resolved_expr = _resolve_expr_columns(df, expr)
try:
mask = df.eval(resolved_expr)
return df[mask].copy()
except Exception as e:
print(f"Warning: Expression '{resolved_expr}' failed: {e}", file=sys.stderr)
return df
def _resolve_expr_columns(df, expr):
"""Replace enum-style field names in expression with actual DataFrame column names."""
# Find all potential column references (uppercase words with underscores)
# Sort by length descending to avoid partial replacements
# e.g., AVERAGE_VOLUME_30_DAY must be replaced before VOLUME
tokens = sorted(set(re.findall(r"[A-Z][A-Z0-9_]+", expr)), key=len, reverse=True)
for token in tokens:
col = find_df_column(df, token)
if col and col != token:
# Use word-boundary replacement to avoid partial matches
expr = re.sub(r"\b" + re.escape(token) + r"\b", f"`{col}`", expr)
return expr
def apply_range(df, config):
"""Filter rows where field is between min and max (inclusive).
Config keys: field (column), min (number), max (number)
"""
col = find_df_column(df, config["field"])
min_val = config["min"]
max_val = config["max"]
if not col:
print(f"Warning: Range field '{config['field']}' not in DataFrame. "
f"Available: {list(df.columns)}", file=sys.stderr)
return df
mask = (df[col] >= min_val) & (df[col] <= max_val)
return df[mask].copy()
# Dispatcher mapping signal type string to handler function
SIGNAL_TYPE_MAP = {
"crossover": apply_crossover,
"threshold": apply_threshold,
"expression": apply_expression,
"range": apply_range,
}
def apply_computed_signals(df, computed_list):
"""Apply a list of computed signal configs to a DataFrame (AND chain).
Args:
df: Input DataFrame from tvscreener
computed_list: List of signal config dicts, each with 'type' key
Returns:
Filtered DataFrame after all signals applied
"""
if not computed_list:
return df
total_before = len(df)
for signal in computed_list:
sig_type = signal.get("type")
if sig_type not in SIGNAL_TYPE_MAP:
raise ValueError(
f"Unknown signal type '{sig_type}'. "
f"Available: {list(SIGNAL_TYPE_MAP.keys())}"
)
handler = SIGNAL_TYPE_MAP[sig_type]
df = handler(df, signal)
print(f"Signals: {total_before} → {len(df)} rows after filtering", file=sys.stderr)
return df
```
### scripts/tests/test_screen.py
```python
"""Tests for screen.py - core screener CLI."""
import sys
import os
import pytest
import pandas as pd
from unittest.mock import patch, MagicMock
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from screener_constants import SCREENER_MAP, OP_MAP, resolve_field, find_df_column
from screen import apply_filters, select_columns, format_markdown
# -- Fixtures --
@pytest.fixture
def stock_df():
return pd.DataFrame({
"Symbol": ["NYSE:AAPL", "NYSE:GOOGL", "NYSE:MSFT", "NYSE:TSLA", "NYSE:AMZN"],
"Name": ["AAPL", "GOOGL", "MSFT", "TSLA", "AMZN"],
"Price": [150.0, 140.0, 350.0, 200.0, 170.0],
"Volume": [50e6, 30e6, 25e6, 80e6, 40e6],
"Change %": [2.5, -1.0, 0.5, 5.0, 1.2],
})
# -- SCREENER_MAP tests --
def test_screener_map_has_all_asset_classes():
expected = {"stock", "crypto", "forex", "bond", "futures", "coin"}
assert set(SCREENER_MAP.keys()) == expected
def test_screener_map_values_are_tuples():
for key, val in SCREENER_MAP.items():
assert isinstance(val, tuple) and len(val) == 2, f"{key} should map to (Screener, Field) tuple"
# -- OP_MAP tests --
def test_op_map_has_all_operators():
expected = {">", ">=", "<", "<=", "==", "!=", "between", "isin"}
assert set(OP_MAP.keys()) == expected
# -- resolve_field tests --
def test_resolve_field_valid():
import tvscreener as tvs
field = resolve_field(tvs.StockField, "PRICE")
assert field == tvs.StockField.PRICE
def test_resolve_field_case_insensitive():
import tvscreener as tvs
field = resolve_field(tvs.StockField, "price")
assert field == tvs.StockField.PRICE
def test_resolve_field_invalid_raises():
import tvscreener as tvs
with pytest.raises(ValueError, match="not found"):
resolve_field(tvs.StockField, "NONEXISTENT_FIELD_XYZ")
# -- find_df_column tests --
def test_find_df_column_exact(stock_df):
assert find_df_column(stock_df, "Volume") == "Volume"
def test_find_df_column_case_insensitive(stock_df):
assert find_df_column(stock_df, "VOLUME") == "Volume"
def test_find_df_column_enum_to_display():
df = pd.DataFrame(columns=["Simple Moving Average (50)", "Relative Strength Index (14)"])
assert find_df_column(df, "SIMPLE_MOVING_AVERAGE_50") == "Simple Moving Average (50)"
assert find_df_column(df, "RELATIVE_STRENGTH_INDEX_14") == "Relative Strength Index (14)"
def test_find_df_column_not_found(stock_df):
assert find_df_column(stock_df, "NONEXISTENT") is None
# -- format_markdown tests --
def test_format_markdown_basic(stock_df):
result = format_markdown(stock_df, "stock")
assert "**Stock Screener**" in result
assert "5 of 5 results" in result
def test_format_markdown_with_sort(stock_df):
result = format_markdown(stock_df, "stock", sort_by="VOLUME", sort_order="desc")
assert "Sorted by VOLUME desc" in result
# First data line should have highest volume (TSLA=80M)
lines = result.split("\n")
data_lines = [l for l in lines if "TSLA" in l]
assert len(data_lines) > 0
def test_format_markdown_with_limit(stock_df):
result = format_markdown(stock_df, "stock", limit=2)
assert "2 of 5 results" in result
def test_format_markdown_empty():
df = pd.DataFrame()
result = format_markdown(df, "stock")
assert "0 results" in result
assert "No matches found" in result
# -- apply_filters tests --
def test_apply_filters_none():
screener = MagicMock()
result = apply_filters(screener, None, None)
assert result == screener
screener.where.assert_not_called()
def test_apply_filters_invalid_json(capsys):
screener = MagicMock()
with pytest.raises(SystemExit):
apply_filters(screener, None, "not valid json")
# -- select_columns tests --
def test_select_columns_none():
screener = MagicMock()
result = select_columns(screener, None, None)
assert result == screener
screener.select.assert_not_called()
```
### scripts/tests/test_signal_engine.py
```python
"""Tests for signal_types.py and signal_engine.py."""
import sys
import os
import pytest
import pandas as pd
import tempfile
import yaml
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
from signal_types import (
apply_crossover, apply_threshold, apply_expression, apply_range,
apply_computed_signals, validate_expression, SIGNAL_TYPE_MAP,
)
from signal_engine import load_signal, list_signals
# -- Fixtures --
@pytest.fixture
def crossover_df():
return pd.DataFrame({
"Name": ["A", "B", "C", "D"],
"SMA50": [100, 200, 150, 80],
"SMA200": [95, 210, 148, 85],
"Price": [105, 195, 155, 78],
})
@pytest.fixture
def indicator_df():
return pd.DataFrame({
"Name": ["A", "B", "C", "D", "E"],
"RSI": [25, 45, 70, 15, 55],
"Price": [100, 200, 300, 50, 150],
"Volume": [1e6, 5e6, 2e6, 8e6, 3e6],
"AvgVolume": [0.5e6, 4e6, 3e6, 2e6, 3e6],
"Change": [2.0, -1.0, 0.5, 5.0, 1.2],
})
# -- Signal type tests --
def test_crossover_up(crossover_df):
config = {"fast": "SMA50", "slow": "SMA200", "direction": "up"}
result = apply_crossover(crossover_df, config)
assert len(result) == 2 # A (100>95) and C (150>148)
assert list(result["Name"]) == ["A", "C"]
def test_crossover_down(crossover_df):
config = {"fast": "SMA50", "slow": "SMA200", "direction": "down"}
result = apply_crossover(crossover_df, config)
assert len(result) == 2 # B (200<210) and D (80<85)
assert list(result["Name"]) == ["B", "D"]
def test_crossover_both(crossover_df):
config = {"fast": "SMA50", "slow": "SMA200", "direction": "both"}
result = apply_crossover(crossover_df, config)
assert len(result) == 4 # All have different values
def test_crossover_missing_column(crossover_df):
config = {"fast": "NONEXISTENT", "slow": "SMA200", "direction": "up"}
result = apply_crossover(crossover_df, config)
assert len(result) == len(crossover_df) # Returns unchanged
def test_threshold_less_than(indicator_df):
config = {"field": "RSI", "op": "<", "value": 30}
result = apply_threshold(indicator_df, config)
assert len(result) == 2 # A (25) and D (15)
def test_threshold_greater_than(indicator_df):
config = {"field": "RSI", "op": ">", "value": 50}
result = apply_threshold(indicator_df, config)
assert len(result) == 2 # C (70) and E (55)
def test_threshold_missing_field(indicator_df):
config = {"field": "MISSING", "op": ">", "value": 0}
result = apply_threshold(indicator_df, config)
assert len(result) == len(indicator_df)
def test_threshold_invalid_op(indicator_df):
config = {"field": "RSI", "op": "~=", "value": 30}
with pytest.raises(ValueError, match="Unknown threshold operator"):
apply_threshold(indicator_df, config)
def test_expression_valid(indicator_df):
config = {"expr": "Volume > AvgVolume * 2"}
result = apply_expression(indicator_df, config)
assert len(result) == 1 # D (8M > 2M*2=4M); A is equal, not greater
def test_expression_invalid_returns_original(indicator_df):
config = {"expr": "NONEXISTENT > 0"}
result = apply_expression(indicator_df, config)
assert len(result) == len(indicator_df) # Returns unchanged on error
def test_validate_expression_blocked():
with pytest.raises(ValueError, match="blocked keywords"):
validate_expression("import os")
def test_validate_expression_valid():
validate_expression("PRICE > 50")
validate_expression("VOLUME > AVERAGE_VOLUME * 2")
def test_range_filter(indicator_df):
config = {"field": "RSI", "min": 20, "max": 50}
result = apply_range(indicator_df, config)
assert len(result) == 2 # A (25) and B (45)
def test_range_missing_field(indicator_df):
config = {"field": "MISSING", "min": 0, "max": 100}
result = apply_range(indicator_df, config)
assert len(result) == len(indicator_df)
# -- Dispatcher tests --
def test_signal_type_map_has_all_types():
assert set(SIGNAL_TYPE_MAP.keys()) == {"crossover", "threshold", "expression", "range"}
def test_apply_computed_signals_chain(indicator_df):
computed = [
{"type": "threshold", "field": "RSI", "op": "<", "value": 60},
{"type": "threshold", "field": "Change", "op": ">", "value": 0},
]
result = apply_computed_signals(indicator_df, computed)
# RSI<60: A(25), B(45), D(15), E(55) -> Change>0: A(2), D(5), E(1.2)
assert len(result) == 3
def test_apply_computed_signals_empty(indicator_df):
result = apply_computed_signals(indicator_df, [])
assert len(result) == len(indicator_df)
def test_apply_computed_signals_unknown_type(indicator_df):
with pytest.raises(ValueError, match="Unknown signal type"):
apply_computed_signals(indicator_df, [{"type": "unknown"}])
# -- YAML loading tests --
def test_load_signal_valid():
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"name": "test", "asset_class": "stock", "filters": []}, f)
f.flush()
config = load_signal(f.name)
assert config["name"] == "test"
assert config["asset_class"] == "stock"
os.unlink(f.name)
def test_load_signal_missing_file():
with pytest.raises(FileNotFoundError):
load_signal("/nonexistent/path/signal.yaml")
def test_load_signal_invalid_config():
with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as f:
yaml.dump({"invalid": "config"}, f)
f.flush()
with pytest.raises(ValueError, match="missing"):
load_signal(f.name)
os.unlink(f.name)
def test_load_signal_by_name():
"""Test loading signal by name from default signals dir."""
config = load_signal("golden-cross")
assert config["name"] == "golden-cross"
assert config["asset_class"] == "stock"
def test_list_signals(capsys):
list_signals()
output = capsys.readouterr().out
assert "golden-cross" in output
assert "oversold-bounce" in output
assert "volume-breakout" in output
```