Python量化工具：TradingView-Screener股票数据筛选完全指南

【免费下载链接】TradingView-ScreenerA package that lets you create TradingView screeners in Python项目地址: https://gitcode.com/gh_mirrors/tr/TradingView-Screener

在金融数据分析领域，快速准确地筛选股票数据是制定投资策略的基础。TradingView-Screener作为一款强大的Python量化工具，为开发者和投资者提供了程序化访问全球股票市场数据的能力。本文将从基础查询到高级策略构建，全面介绍如何利用该工具实现高效股票数据筛选与分析，帮助你在量化投资的道路上事半功倍。

安装与环境配置：5分钟快速上手

TradingView-Screener的安装过程简单高效，支持通过pip或Poetry两种方式进行安装，满足不同开发环境需求。

基础安装命令

# 使用pip安装 pip install tradingview-screener # 或通过源码安装 git clone https://gitcode.com/gh_mirrors/tr/TradingView-Screener cd TradingView-Screener poetry install

核心模块解析

⚠️ 注意：安装前请确保Python版本≥3.8，并已安装pandas和requests依赖库。

基础查询构建：从简单筛选到多条件组合

初始化查询对象

创建基础查询对象是所有筛选操作的起点，默认配置下将返回美国市场的股票数据：

from tradingview_screener import Query # 创建基础查询 basic_query = Query() # 获取默认筛选结果 total_count, data_frame = basic_query.get_scanner_data() print(f"找到 {total_count} 条记录") print(data_frame.head())

自定义筛选列

使用select()方法指定需要返回的数据列，支持技术指标、基本面数据等250+种列：

# 选择特定列 custom_columns_query = ( Query() .select('name', 'close', 'volume', 'MACD.macd', 'RSI', 'Price to Earnings Ratio (TTM)') ) # 获取筛选结果 _, df = custom_columns_query.get_scanner_data() print(df.columns) # 查看选择的列

条件筛选实现

通过where()方法添加筛选条件，支持多种比较运算符和逻辑组合：

from tradingview_screener.column import Column as col # 创建价格和成交量筛选条件 filtered_query = ( Query() .select('name', 'close', 'volume', 'market_cap_basic') .where( col('close') > 50, # 收盘价大于50 col('volume') > 1_000_000, # 成交量大于100万 col('market_cap_basic').between(1_000_000_000, 10_000_000_000) # 市值在10亿到100亿之间 ) ) _, df = filtered_query.get_scanner_data() print(df)

高级查询技巧：逻辑组合与市场定位

复杂逻辑条件构建

使用And()和Or()函数创建多条件逻辑组合，实现复杂筛选策略：

from tradingview_screener.query import And, Or # 创建复杂逻辑条件 complex_query = ( Query() .select('name', 'close', 'EMA5', 'EMA20', 'type') .where2( And( Or( col('type') == 'stock', col('type') == 'fund' ), col('close').between(col('EMA5'), col('EMA20')) # 收盘价在EMA5和EMA20之间 ) ) ) _, df = complex_query.get_scanner_data() print(df)

市场与资产类型选择

使用set_markets()方法指定目标市场，支持全球67个国家和多种资产类别：

# 多市场筛选 multi_market_query = ( Query() .select('name', 'close', 'market', 'country', 'currency') .set_markets('america', 'europe', 'asia') # 同时筛选美洲、欧洲和亚洲市场 ) _, df = multi_market_query.get_scanner_data() print(df[['name', 'market', 'country']].head(10))

指数成分股筛选

通过set_index()方法筛选特定指数成分股，如纳斯达克100、标普500等：

# 筛选纳斯达克100成分股 nasdaq_query = ( Query() .select('name', 'close', 'market_cap_basic') .set_index('SYML:NASDAQ;NDX') # 纳斯达克100指数 .order_by('market_cap_basic', ascending=False) # 按市值降序排列 ) _, df = nasdaq_query.get_scanner_data() print("纳斯达克100成分股前10名:") print(df.head(10))

数据处理与可视化：从原始数据到决策支持

数据持久化存储

将筛选结果保存为CSV或Excel格式，便于后续分析：

# 保存筛选结果 _, df = nasdaq_query.get_scanner_data() df.to_csv('nasdaq_100_top.csv', index=False) df.to_excel('nasdaq_100_top.xlsx', index=False)

技术指标可视化

结合Matplotlib绘制股票K线图和技术指标，直观展示市场趋势：

import matplotlib.pyplot as plt import mplfinance as mpf # 获取单只股票数据 stock_data = ( Query() .select('open', 'high', 'low', 'close', 'volume') .set_tickers('NASDAQ:NVDA') # NVIDIA股票 .get_scanner_data()[1] ) # 转换为时间序列并绘图 stock_data.index = pd.date_range(start='2023-01-01', periods=len(stock_data)) mpf.plot( stock_data, type='candle', volume=True, title='NVIDIA Stock Price', ylabel='Price ($)', ylabel_lower='Volume', figratio=(12,6) ) plt.savefig('nvda_stock_chart.png')

策略回测模板

构建简单的均值回归策略回测框架，验证策略有效性：

def mean_reversion_strategy(data): """均值回归策略实现""" # 计算20日移动平均线 data['MA20'] = data['close'].rolling(window=20).mean() # 计算价格与均线的偏差百分比 data['deviation'] = (data['close'] - data['MA20']) / data['MA20'] * 100 # 生成交易信号：低于均线2%买入，高于均线2%卖出 data['signal'] = 0 data.loc[data['deviation'] < -2, 'signal'] = 1 # 买入信号 data.loc[data['deviation'] > 2, 'signal'] = -1 # 卖出信号 return data # 获取测试数据 _, test_data = ( Query() .select('close') .set_tickers('NASDAQ:AAPL') .get_scanner_data() ) # 应用策略 result = mean_reversion_strategy(test_data) print(result[['close', 'MA20', 'deviation', 'signal']].tail(10))

性能优化与生态整合：提升筛选效率与扩展能力

分页查询实现

处理大量数据时，使用分页查询提升性能：

def paginated_query(query, page_size=100): """分页获取数据""" all_data = [] offset = 0 while True: _, df = query.offset(offset).limit(page_size).get_scanner_data() if len(df) == 0: break all_data.append(df) offset += page_size print(f"已获取 {offset} 条数据...") return pd.concat(all_data, ignore_index=True) # 创建分页查询 large_query = ( Query() .select('name', 'close', 'volume') .where(col('volume') > 5_000_000) # 成交量大于500万 ) # 分页获取所有结果 all_results = paginated_query(large_query) print(f"总共获取 {len(all_results)} 条记录")

多线程查询加速

使用多线程并行执行多个查询，减少总等待时间：

import concurrent.futures def fetch_market_data(market): """获取单个市场数据""" return ( Query() .select('name', 'close', 'market') .set_markets(market) .limit(50) .get_scanner_data()[1] ) # 多市场并行查询 markets = ['america', 'europe', 'asia', 'japan', 'australia'] with concurrent.futures.ThreadPoolExecutor() as executor: results = list(executor.map(fetch_market_data, markets)) # 合并结果 combined_data = pd.concat(results, ignore_index=True) print(combined_data.groupby('market').size())

与量化生态工具集成

将TradingView-Screener与其他量化工具结合，构建完整分析流程：

# 与TA-Lib技术分析库集成 import talib # 获取原始数据 _, df = ( Query() .select('open', 'high', 'low', 'close', 'volume') .set_tickers('NASDAQ:MSFT') .get_scanner_data() ) # 计算技术指标 df['RSI'] = talib.RSI(df['close'].values, timeperiod=14) df['MACD'], df['MACD_signal'], df['MACD_hist'] = talib.MACD( df['close'].values, fastperiod=12, slowperiod=26, signalperiod=9 ) print(df[['close', 'RSI', 'MACD', 'MACD_signal']].tail(10))

常见问题解决：从调试到部署

API请求错误处理

添加异常处理机制，确保程序稳定运行：

import requests def safe_query(query, max_retries=3): """带重试机制的安全查询""" retries = 0 while retries < max_retries: try: return query.get_scanner_data() except requests.exceptions.RequestException as e: retries += 1 print(f"请求失败 (重试 {retries}/{max_retries}): {e}") if retries == max_retries: raise time.sleep(2 ** retries) # 指数退避策略 # 使用安全查询 safe_query(Query().select('close').limit(10))

常见错误及解决方案

生产环境部署建议

1. 请求频率控制：将API调用频率控制在每分钟10次以内，避免触发限制
2. 缓存机制：对频繁查询的静态数据进行缓存，减少重复请求
3. 日志记录：实现详细日志记录，便于问题排查
4. 监控告警：设置关键指标监控，异常时及时告警

# 生产环境查询模板 import logging from time import sleep # 配置日志 logging.basicConfig(filename='screener.log', level=logging.INFO) def production_query(query, query_name, cache_ttl=3600): """生产环境查询函数""" # 检查缓存 cache_file = f"cache/{query_name}.pkl" if os.path.exists(cache_file) and time.time() - os.path.getmtime(cache_file) < cache_ttl: logging.info(f"使用缓存: {query_name}") return pd.read_pickle(cache_file) # 执行查询 try: logging.info(f"执行查询: {query_name}") count, df = safe_query(query) # 保存缓存 os.makedirs("cache", exist_ok=True) df.to_pickle(cache_file) return df except Exception as e: logging.error(f"查询失败: {query_name}, 错误: {str(e)}") raise

总结与扩展：构建专业量化系统

TradingView-Screener作为一款功能强大的Python量化工具，为股票数据筛选提供了灵活高效的解决方案。通过本文介绍的基础查询、高级筛选、数据可视化和性能优化技巧，你可以构建从简单筛选到复杂策略回测的完整量化分析流程。

未来扩展方向：

1. 结合机器学习模型进行股价预测
2. 构建实时监控与自动交易系统
3. 多因子选股策略开发
4. 跨境市场 arbitrage 机会挖掘

无论你是量化交易新手还是经验丰富的开发者，TradingView-Screener都能为你提供强大的数据支持，帮助你在金融市场中发现价值机会。通过不断探索和实践，你可以将这些工具和技巧转化为切实可行的投资策略，在量化投资的道路上稳步前行。

【免费下载链接】TradingView-ScreenerA package that lets you create TradingView screeners in Python项目地址: https://gitcode.com/gh_mirrors/tr/TradingView-Screener