TradingView股票筛选器终极指南:用Python实现专业级量化分析
TradingView股票筛选器终极指南:用Python实现专业级量化分析
TradingView-Screener是一个功能强大的Python包,专门用于创建TradingView股票筛选器,让开发者能够通过编程方式快速获取和分析全球金融市场数据。这个工具为量化交易员、数据分析师和金融开发者提供了直接访问TradingView官方API的能力,无需网页爬取或HTML解析。无论你是构建自动化交易系统、进行市场研究还是开发金融分析工具,TradingView-Screener都能显著提升你的工作效率。
项目概述与核心价值
TradingView-Screener的核心优势在于其简洁的API设计和强大的数据获取能力。作为TradingView官方API的Python封装,它提供了超过3000个数据字段的访问权限,涵盖股票、期权、加密货币、外汇、期货、债券等多种金融工具。这个工具特别适合需要批量处理市场数据、构建自定义筛选策略或进行实时市场监控的开发者和分析师。
核心功能亮点
- 多市场支持:覆盖全球70多个国家的股票市场,以及加密货币、外汇、期货、债券等金融工具
- 丰富数据字段:提供3000+数据字段,包括OHLC价格数据、技术指标、基本面指标和TradingView专有字段
- 灵活时间框架:支持1分钟到1个月的时间框架,可以自由混合不同时间框架的字段
- SQL式筛选语法:支持完整的AND/OR逻辑运算,让筛选条件编写更加直观
- 实时数据访问:通过会话认证支持实时数据流访问
快速开始指南
环境安装与配置
开始使用TradingView-Screener非常简单,只需几个步骤即可完成环境搭建:
# 安装包
pip install tradingview-screener
# 可选:安装rookiepy用于自动加载浏览器cookies
pip install rookiepy
基础使用示例
以下是一个简单的入门示例,展示如何获取美国市场的前50支股票数据:
from tradingview_screener import Query
# 创建查询实例并获取数据
result_count, data_df = (Query()
.select('name', 'close', 'volume', 'market_cap_basic')
.get_scanner_data())
print(f"总记录数: {result_count}")
print(data_df.head())
这个简单的代码片段将返回包含股票名称、收盘价、成交量和市值的DataFrame,默认限制为50行数据。
核心功能详解
多市场数据获取
TradingView-Screener支持多种金融市场的数据查询:
from tradingview_screener import stocks, crypto, options, forex, futures
# 获取意大利市值最高的5支股票
italian_stocks = stocks('italy').limit(5).get_scanner_data()
# 获取交易量最高的5个中心化交易所加密货币对
top_crypto = crypto().limit(5).get_scanner_data()
# 获取AAPL期权链数据
aapl_options = options('NASDAQ:AAPL').limit(5).get_scanner_data()
# 获取外汇主要货币对
major_forex = forex().limit(5).get_scanner_data()
# 获取期货合约数据
commodity_futures = futures().limit(5).get_scanner_data()
高级筛选功能
通过灵活的筛选条件,你可以构建复杂的查询逻辑:
from tradingview_screener import Query, col
# 构建复杂筛选条件
advanced_query = (Query()
.select('name', 'close', 'close|1', 'close|5', 'volume', 'relative_volume_10d_calc')
.where(
col('market_cap_basic').between(1_000_000, 50_000_000),
col('relative_volume_10d_calc') > 1.2,
col('MACD.macd|1') >= col('MACD.signal|1') # 1分钟MACD指标
)
.order_by('volume', ascending=False)
.offset(5)
.limit(25)
.get_scanner_data())
时间框架支持
TradingView-Screener支持多种时间框架,可以混合使用:
| 时间框架 | 列名示例 | 说明 |
|---|---|---|
| 1分钟 | close\|1 |
1分钟收盘价 |
| 5分钟 | close\|5 |
5分钟收盘价 |
| 15分钟 | close\|15 |
15分钟收盘价 |
| 30分钟 | close\|30 |
30分钟收盘价 |
| 1小时 | close\|60 |
1小时收盘价 |
| 4小时 | close\|240 |
4小时收盘价 |
| 1日 | close |
日线收盘价 |
| 1周 | close\|1W |
周线收盘价 |
| 1月 | close\|1M |
月线收盘价 |
实战应用案例
案例一:开盘前策略准备
对于日内交易者,开盘前的市场分析至关重要。以下代码展示了如何筛选出预市表现活跃的股票:
from tradingview_screener import Query, col
# 筛选预市涨幅超过5%且成交量放大的股票
premarket_strategy = (Query()
.select('name', 'close', 'premarket_change', 'volume', 'relative_volume_10d_calc')
.where(
col('premarket_change') > 5.0, # 预市涨幅超过5%
col('relative_volume_10d_calc') > 1.5, # 相对成交量放大50%
col('market_cap_basic') > 1_000_000_000 # 市值大于10亿美元
)
.order_by('premarket_change', ascending=False)
.limit(20)
.get_scanner_data())
案例二:技术指标筛选策略
结合多个技术指标构建量化筛选策略:
from tradingview_screener import Query, col
# 多指标综合筛选
technical_screening = (Query()
.select('name', 'close', 'RSI', 'MACD.macd', 'MACD.signal', 'volume')
.where(
col('RSI') < 30, # RSI超卖
col('MACD.macd') > col('MACD.signal'), # MACD金叉
col('volume') > 1_000_000, # 成交量充足
col('close') > 10.0 # 股价高于10美元
)
.order_by('RSI', ascending=True)
.limit(15)
.get_scanner_data())
案例三:投资组合监控
定期监控持仓股票的表现和风险指标:
import pandas as pd
from tradingview_screener import Query
# 定义持仓股票列表
portfolio_tickers = ['NASDAQ:AAPL', 'NASDAQ:GOOGL', 'NASDAQ:MSFT', 'NASDAQ:AMZN', 'NASDAQ:TSLA']
# 批量获取持仓股票数据
portfolio_data = []
for ticker in portfolio_tickers:
stock_data = (Query()
.select('name', 'close', 'change', 'volume', 'market_cap_basic', 'beta')
.where(col('ticker') == ticker)
.get_scanner_data())
portfolio_data.append(stock_data[1])
# 合并分析
portfolio_df = pd.concat(portfolio_data, ignore_index=True)
print(portfolio_df)
高级配置与优化
实时数据访问配置
要访问实时数据,需要配置会话cookies。以下是使用rookiepy自动加载浏览器cookies的方法:
import rookiepy
from tradingview_screener import Query
# 从Chrome浏览器加载cookies
cookies = rookiepy.to_cookiejar(rookiepy.chrome(['.tradingview.com']))
# 使用cookies获取实时数据
real_time_data = Query().select('name', 'close', 'bid', 'ask').get_scanner_data(cookies=cookies)
# 验证数据更新模式
_, df = Query().select('exchange', 'update_mode').limit(1_000_000).get_scanner_data(cookies=cookies)
update_stats = df.groupby('exchange')['update_mode'].value_counts()
print(update_stats)
手动Cookie配置
如果无法使用rookiepy,可以手动提取cookies:
# 手动配置cookies
cookies = {'sessionid': 'your-session-id-here'}
# 使用手动cookies查询
data = Query().get_scanner_data(cookies=cookies)
API请求优化
对于大规模数据查询,建议进行性能优化:
from tradingview_screener import Query
import time
def batch_query_with_delay(tickers, batch_size=10, delay_seconds=2):
"""批量查询带延迟,避免API限制"""
results = []
for i in range(0, len(tickers), batch_size):
batch = tickers[i:i+batch_size]
query = Query().where(col('ticker').isin(batch))
results.append(query.get_scanner_data())
time.sleep(delay_seconds) # 添加延迟避免请求过快
return results
# 使用批量查询
ticker_list = ['NASDAQ:AAPL', 'NASDAQ:GOOGL', 'NASDAQ:MSFT', 'NASDAQ:AMZN']
batch_results = batch_query_with_delay(ticker_list)
生态系统集成
与Pandas数据分析集成
TradingView-Screener返回的数据可以直接转换为Pandas DataFrame,便于进行进一步的数据分析:
import pandas as pd
from tradingview_screener import Query
# 获取数据并转换为DataFrame
result_count, df = (Query()
.select('name', 'close', 'volume', 'market_cap_basic', 'pe')
.limit(100)
.get_scanner_data())
# 数据清洗和分析
df_clean = df.dropna(subset=['pe', 'market_cap_basic'])
df_clean['pe_ratio'] = df_clean['pe']
df_clean['market_cap'] = df_clean['market_cap_basic']
# 计算统计指标
summary_stats = df_clean.describe()
correlation_matrix = df_clean[['close', 'volume', 'market_cap', 'pe_ratio']].corr()
# 导出数据
df_clean.to_csv('stock_analysis.csv', index=False)
与Matplotlib可视化集成
结合数据可视化工具创建专业的分析图表:
import matplotlib.pyplot as plt
import pandas as pd
from tradingview_screener import Query
# 获取数据
_, df = (Query()
.select('name', 'close', 'volume', 'market_cap_basic')
.order_by('market_cap_basic', ascending=False)
.limit(20)
.get_scanner_data())
# 创建可视化图表
fig, axes = plt.subplots(2, 2, figsize=(12, 10))
# 市值分布图
axes[0, 0].bar(df['name'], df['market_cap_basic'] / 1e9)
axes[0, 0].set_title('Top 20 Stocks by Market Cap (Billions)')
axes[0, 0].tick_params(axis='x', rotation=45)
# 价格与成交量散点图
axes[0, 1].scatter(df['close'], df['volume'], alpha=0.6)
axes[0, 1].set_xlabel('Price')
axes[0, 1].set_ylabel('Volume')
axes[0, 1].set_title('Price vs Volume')
# 市值与价格关系
axes[1, 0].scatter(df['market_cap_basic'] / 1e9, df['close'], alpha=0.6)
axes[1, 0].set_xlabel('Market Cap (Billions)')
axes[1, 0].set_ylabel('Price')
axes[1, 0].set_title('Market Cap vs Price')
# 成交量分布直方图
axes[1, 1].hist(df['volume'], bins=20, alpha=0.7)
axes[1, 1].set_xlabel('Volume')
axes[1, 1].set_ylabel('Frequency')
axes[1, 1].set_title('Volume Distribution')
plt.tight_layout()
plt.savefig('stock_analysis.png', dpi=300, bbox_inches='tight')
plt.show()
最佳实践与建议
错误处理与重试机制
在实际应用中,建议添加完善的错误处理机制:
import time
from tradingview_screener import Query
import requests
def safe_get_data(query, max_retries=3, retry_delay=5):
"""安全的获取数据函数,包含重试机制"""
for attempt in range(max_retries):
try:
result_count, data_df = query.get_scanner_data()
return result_count, data_df
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
print(f"请求失败,{retry_delay}秒后重试... (尝试 {attempt + 1}/{max_retries})")
time.sleep(retry_delay)
else:
raise Exception(f"获取数据失败,已达最大重试次数: {e}")
except Exception as e:
raise Exception(f"获取数据时发生错误: {e}")
# 使用安全函数
try:
result = safe_get_data(Query().select('name', 'close').limit(10))
print(f"成功获取{result[0]}条记录")
except Exception as e:
print(f"错误: {e}")
性能优化建议
- 批量查询优化:避免频繁的小规模查询,尽量合并查询条件
- 数据缓存策略:对不常变化的数据实施缓存机制
- 请求频率控制:合理设置请求间隔,避免触发API限制
- 选择性字段获取:只获取需要的字段,减少数据传输量
from tradingview_screener import Query
import hashlib
import pickle
import os
class CachedQuery:
"""带缓存的查询类"""
def __init__(self, cache_dir='./cache', ttl_seconds=300):
self.cache_dir = cache_dir
self.ttl_seconds = ttl_seconds
os.makedirs(cache_dir, exist_ok=True)
def get_cached_data(self, query, cache_key):
"""获取缓存数据"""
cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
if os.path.exists(cache_file):
file_age = time.time() - os.path.getmtime(cache_file)
if file_age < self.ttl_seconds:
with open(cache_file, 'rb') as f:
return pickle.load(f)
return None
def set_cached_data(self, data, cache_key):
"""设置缓存数据"""
cache_file = os.path.join(self.cache_dir, f"{cache_key}.pkl")
with open(cache_file, 'wb') as f:
pickle.dump(data, f)
def execute_with_cache(self, query):
"""执行带缓存的查询"""
# 生成缓存键
query_str = str(query.__dict__)
cache_key = hashlib.md5(query_str.encode()).hexdigest()
# 尝试从缓存获取
cached_result = self.get_cached_data(query, cache_key)
if cached_result is not None:
print("从缓存加载数据")
return cached_result
# 执行查询并缓存结果
result = query.get_scanner_data()
self.set_cached_data(result, cache_key)
print("执行新查询并缓存结果")
return result
# 使用缓存查询
cached_query = CachedQuery()
result = cached_query.execute_with_cache(Query().select('name', 'close').limit(10))
项目结构参考
了解项目结构有助于更好地使用和扩展TradingView-Screener:
TradingView-Screener/
├── src/
│ └── tradingview_screener/
│ ├── __init__.py # 包入口和主要API
│ ├── column.py # 列定义和筛选条件
│ ├── models.py # 数据模型
│ ├── query.py # 查询构建器
│ ├── screeners.py # 筛选器实现
│ └── util.py # 工具函数
├── tests/ # 测试用例
│ ├── test_query.py
│ ├── test_readme.py
│ └── test_screeners.py
├── pyproject.toml # 项目配置
└── README.md # 项目文档
通过合理使用TradingView-Screener,你可以构建强大的金融数据分析应用,实现自动化交易策略回测、实时市场监控和投资组合分析。这个工具为Python开发者提供了直接访问TradingView丰富数据资源的桥梁,极大地简化了金融数据获取和分析的复杂度。
记住,虽然工具能提供强大的数据支持,但成功的投资决策还需要结合深入的市场理解、严格的风险管理和持续的学习。希望这份指南能帮助你在量化分析和自动化交易的道路上取得更大的成功!
更多推荐


所有评论(0)