Python金融数据获取终极指南使用pywencai高效访问同花顺问财数据【免费下载链接】pywencai获取同花顺问财数据项目地址: https://gitcode.com/gh_mirrors/py/pywencai在金融数据分析和量化投资领域获取高质量、实时的市场数据是每个数据分析师和开发者面临的核心挑战。传统方法如网页爬虫开发复杂、维护成本高而官方API往往价格昂贵或功能有限。pywencai作为一个开源Python工具为开发者提供了高效、稳定的同花顺问财数据访问方案让金融数据获取变得简单可靠。项目价值主张为什么选择pywencaipywencai的核心价值在于其智能化的一站式金融数据获取能力。通过模拟浏览器行为并解析同花顺问财接口该工具将复杂的网页数据提取过程封装为简洁的Python API让开发者能够专注于数据分析而非数据获取的技术细节。核心优势对比特性维度pywencai解决方案传统网页爬虫官方API接口开发复杂度极低几行代码即可使用高需要处理反爬、解析等中等需要学习API文档数据完整性完整支持所有问财功能有限依赖网页结构完整但可能有功能限制稳定性保障内置重试机制和错误处理脆弱易受网站变更影响稳定但可能有调用限制成本效益完全免费开源免费但维护成本高通常需要付费实时性接近实时与网页同步实时但易被封禁实时但有速率限制扩展性易于集成到现有系统集成复杂标准API集成技术架构创新pywencai采用模块化设计核心架构包含三个关键组件请求引擎(wencai.py) - 处理HTTP通信和重试逻辑数据转换器(convert.py) - 解析JSON响应为DataFrame请求头生成器(headers.py) - 动态生成合法请求头这种分层架构确保了代码的可维护性和扩展性每个模块都专注于单一职责。核心功能演示5分钟快速上手环境配置与安装首先确保系统已安装Node.js v16这是执行JavaScript代码生成请求令牌的前提# 安装pywencai pip install pywencai # 验证安装 python -c import pywencai; print(安装成功)获取Cookie凭证访问同花顺问财网站获取Cookie是使用pywencai的关键步骤使用Chrome浏览器访问www.iwencai.com按F12打开开发者工具切换到Network标签页刷新页面并选择任意POST请求在请求头中找到Cookie字段并复制完整值图1浏览器开发者工具中获取Cookie的详细步骤红色箭头标注关键Cookie字段位置基础数据查询示例import pywencai import pandas as pd # 基础查询获取沪深300成分股中市盈率低于30的股票 df pywencai.get( query沪深300成分股 市盈率30, cookieyour_cookie_value_here, # 替换为实际Cookie loopTrue, # 自动分页获取全部数据 perpage100, # 每页数据量 sort_key市盈率, # 排序字段 sort_orderasc # 升序排列 ) print(f获取到 {len(df)} 条数据) print(df[[股票代码, 股票名称, 市盈率, 总市值]].head()) # 保存到CSV文件 df.to_csv(low_pe_stocks.csv, indexFalse, encodingutf-8-sig)高级查询功能# 多条件组合查询 complex_query pywencai.get( query连续3年ROE15% 资产负债率50% 市值200亿, cookieyour_cookie_value, query_typestock, loopTrue, logTrue # 启用日志输出 ) # 特定股票查询 specific_stocks pywencai.get( query股票, cookieyour_cookie_value, find[600519, 000858, 000001], # 贵州茅台、五粮液、平安银行 perpage50 )架构设计解析深入理解技术实现请求流程架构核心模块详解1. 请求头生成机制 (headers.py)# headers.py核心代码片段 def get_token(): 获取hexin-v令牌 result subprocess.run( [node, os.path.join(os.path.dirname(__file__), hexin-v.bundle.js)], stdoutsubprocess.PIPE ) return result.stdout.decode().strip() def headers(cookieNone, user_agentNone): 构建请求头 if user_agent is None: from fake_useragent import UserAgent ua UserAgent() user_agent ua.random return { hexin-v: get_token(), # 动态生成的加密令牌 User-Agent: user_agent, # 随机用户代理 cookie: cookie # 用户提供的Cookie }该模块通过执行JavaScript代码生成hexin-v令牌这是同花顺问财接口的重要验证参数。使用Node.js执行预编译的JavaScript文件确保令牌生成的准确性和实时性。2. 智能重试机制 (wencai.py)def while_do(do, retry10, sleep0, logFalse): 智能重试装饰器 count 0 while count retry: time.sleep(sleep) try: return do() except: log and logger.warning(f{count1}次尝试失败) count 1 return None重试机制支持指数退避策略在网络不稳定或接口临时限制时自动重试默认最多10次确保数据获取的可靠性。3. 数据转换引擎 (convert.py)def convert(res): 处理API响应数据 result json.loads(res.text) content _.get(result, data.answer.0.txt.0.content) if type(content) str: content json.loads(content) components content[components] # 根据组件类型选择不同处理器 if (len(components) 1 and _.get(components[0], show_type) xuangu_tableV1): # 处理选股表格数据 return xuangu_tableV1_handler(components[0], components) else: # 处理复杂数据结构 return multi_show_type_handler(components)数据转换器支持多种数据类型包括表格数据、文本数据、嵌套结构等确保返回统一的数据格式。配置文件说明项目的配置文件结构清晰便于维护pyproject.toml- Python项目配置和依赖管理package.json- Node.js依赖管理包含jsdom等JavaScript运行时依赖webpack.config.js- JavaScript打包配置用于生成hexin-v.bundle.js实战应用场景金融数据分析案例场景1价值投资筛选系统class ValueInvestmentScreener: def __init__(self, cookie): self.cookie cookie def screen_by_factors(self, factors_config): 基于多因子进行股票筛选 factors_config: 因子配置字典 results {} for factor_name, query in factors_config.items(): try: df pywencai.get( queryquery, cookieself.cookie, loopTrue, logFalse, retry5, sleep0.5 # 避免请求过于频繁 ) if not df.empty: # 数据清洗和验证 df self._clean_data(df) results[factor_name] df print(f{factor_name}因子筛选完成: {len(df)}只股票) else: print(f{factor_name}因子未找到匹配股票) except Exception as e: print(f{factor_name}因子筛选失败: {str(e)}) continue return results def _clean_data(self, df): 数据清洗和标准化 # 去除重复数据 df df.drop_duplicates(subset[股票代码]) # 处理缺失值 numeric_columns [市盈率, 市净率, ROE, 净利润增长率] for col in numeric_columns: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) return df # 使用示例 screener ValueInvestmentScreener(cookieyour_cookie) factors { 高ROE低负债: 连续3年ROE15% 资产负债率50%, 低估值高成长: 市盈率20 净利润增长率30%, 高股息率: 股息率3% 连续3年分红 } results screener.screen_by_factors(factors)场景2技术指标监控系统import schedule import time from datetime import datetime class TechnicalIndicatorMonitor: def __init__(self, cookie, indicators): self.cookie cookie self.indicators indicators self.history_data {} def monitor_market_signals(self): 监控市场技术信号 signals {} for indicator_name, query in self.indicators.items(): try: df pywencai.get( queryquery, cookieself.cookie, loopFalse, # 实时监控不需要分页 perpage20, # 只获取前20条 sort_key涨幅, sort_orderdesc ) if not df.empty: signals[indicator_name] { count: len(df), top_stocks: df[[股票代码, 股票名称, 涨幅]].head(5).to_dict(records), timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } except Exception as e: print(f指标{indicator_name}监控失败: {str(e)}) return signals def start_daily_monitoring(self): 启动每日监控任务 schedule.every().day.at(09:30).do(self.daily_morning_report) schedule.every().day.at(15:00).do(self.daily_closing_report) while True: schedule.run_pending() time.sleep(60) # 技术指标配置 indicators { MACD金叉: MACD金叉 成交量放大, 突破均线: 股价站上20日均线 成交量5日均量, 强势反弹: 涨幅5% 换手率3% } monitor TechnicalIndicatorMonitor(cookieyour_cookie, indicatorsindicators)场景3行业数据分析报告import pandas as pd import matplotlib.pyplot as plt class IndustryAnalysis: def __init__(self, cookie): self.cookie cookie def analyze_industry_trend(self, industries, metrics): 分析多个行业的数据趋势 industries: 行业列表 metrics: 分析指标列表 industry_data {} for industry in industries: print(f正在分析{industry}行业...) # 构建行业查询 query f{industry}行业 {metrics} try: df pywencai.get( queryquery, cookieself.cookie, loopTrue, query_typestock, sleep1 # 行业间请求间隔 ) if not df.empty: # 数据预处理 df self._preprocess_industry_data(df, metrics) industry_data[industry] df # 生成行业统计 stats self._calculate_industry_stats(df, metrics) print(f{industry}行业分析完成: {stats}) except Exception as e: print(f{industry}行业分析失败: {str(e)}) continue return industry_data def generate_comparison_report(self, industry_data): 生成行业对比报告 comparison_df pd.DataFrame() for industry, df in industry_data.items(): if not df.empty: # 计算行业平均指标 avg_values {} numeric_cols df.select_dtypes(include[float64, int64]).columns for col in numeric_cols: if col in df.columns: avg_values[f{industry}_{col}_avg] df[col].mean() comparison_df pd.concat([comparison_df, pd.DataFrame([avg_values])], ignore_indexTrue) return comparison_df # 使用示例 analyzer IndustryAnalysis(cookieyour_cookie) industries [新能源, 人工智能, 生物医药, 半导体] metrics 市盈率 市净率 ROE 营业收入增长率 industry_data analyzer.analyze_industry_trend(industries, metrics) report analyzer.generate_comparison_report(industry_data)生态整合方案与企业级系统对接与数据库系统集成import sqlalchemy as sa from sqlalchemy import create_engine, Table, Column, Integer, String, Float, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base declarative_base() class StockData(Base): __tablename__ stock_data id Column(Integer, primary_keyTrue) stock_code Column(String(10), indexTrue) stock_name Column(String(50)) pe_ratio Column(Float) pb_ratio Column(Float) roe Column(Float) market_cap Column(Float) query_date Column(DateTime) created_at Column(DateTime, defaultdatetime.now) class DataPipeline: def __init__(self, cookie, db_url): self.cookie cookie self.engine create_engine(db_url) Base.metadata.create_all(self.engine) self.Session sessionmaker(bindself.engine) def daily_data_collection(self, queries): 每日数据收集管道 session self.Session() for query_name, query_str in queries.items(): try: df pywencai.get( queryquery_str, cookieself.cookie, loopTrue, sleep0.5 ) if not df.empty: # 数据转换和存储 self._save_to_database(session, df, query_name) print(f{query_name}数据收集完成: {len(df)}条记录) except Exception as e: print(f{query_name}数据收集失败: {str(e)}) session.rollback() continue session.commit() session.close() def _save_to_database(self, session, df, query_name): 保存数据到数据库 for _, row in df.iterrows(): stock_data StockData( stock_coderow.get(股票代码, ), stock_namerow.get(股票名称, ), pe_ratioself._safe_float(row.get(市盈率)), pb_ratioself._safe_float(row.get(市净率)), roeself._safe_float(row.get(ROE)), market_capself._safe_float(row.get(总市值)), query_datedatetime.now() ) session.add(stock_data) def _safe_float(self, value): 安全转换为浮点数 try: return float(value) if value else None except (ValueError, TypeError): return None # 配置数据库连接 pipeline DataPipeline( cookieyour_cookie, db_urlpostgresql://user:passwordlocalhost:5432/finance_db ) # 定义数据收集任务 daily_queries { 沪深300成分股: 沪深300成分股, 高增长股票: 净利润增长率30% 市值100亿, 低估值股票: 市盈率15 市净率2 } pipeline.daily_data_collection(daily_queries)与消息队列系统集成import redis import json from datetime import datetime class RealTimeDataStream: def __init__(self, cookie, redis_hostlocalhost, redis_port6379): self.cookie cookie self.redis_client redis.Redis(hostredis_host, portredis_port, db0) def stream_market_data(self, queries, interval60): 实时数据流推送 import time while True: for stream_name, query in queries.items(): try: # 获取实时数据 df pywencai.get( queryquery, cookieself.cookie, loopFalse, perpage50, sort_key更新时间, sort_orderdesc ) if not df.empty: # 转换为JSON格式 data { stream: stream_name, timestamp: datetime.now().isoformat(), count: len(df), data: df.head(10).to_dict(records) } # 发布到Redis Stream self.redis_client.xadd( market_data_stream, {data: json.dumps(data, ensure_asciiFalse)} ) print(f{stream_name}数据流更新: {len(df)}条记录) except Exception as e: print(f{stream_name}数据流更新失败: {str(e)}) # 等待下一个周期 time.sleep(interval) def get_latest_data(self, stream_name, count10): 获取最新的数据 messages self.redis_client.xrevrange( market_data_stream, , -, countcount ) return [ json.loads(msg[1][bdata].decode(utf-8)) for msg in messages if json.loads(msg[1][bdata].decode(utf-8))[stream] stream_name ] # 实时数据流配置 streamer RealTimeDataStream(cookieyour_cookie) stream_queries { hot_stocks: 涨幅5% 换手率3%, volume_leaders: 成交量排名前50, breakout_stocks: 创60日新高 } # 在后台线程中启动数据流 import threading stream_thread threading.Thread( targetstreamer.stream_market_data, args(stream_queries, 60) # 60秒间隔 ) stream_thread.daemon True stream_thread.start()性能优化与故障排查性能优化技巧批量请求优化# 使用连接池和会话重用 import requests from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry session requests.Session() retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504] ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 在pywencai中使用自定义会话 df pywencai.get( query沪深300, cookieyour_cookie, request_params{session: session} )缓存机制实现import hashlib import pickle from functools import lru_cache import os class CachedWencai: def __init__(self, cookie, cache_dir./cache, ttl3600): self.cookie cookie self.cache_dir cache_dir self.ttl ttl # 缓存有效期秒 if not os.path.exists(cache_dir): os.makedirs(cache_dir) def get_with_cache(self, query, **kwargs): 带缓存的查询 # 生成缓存键 cache_key hashlib.md5( f{query}_{str(kwargs)}.encode() ).hexdigest() cache_file os.path.join(self.cache_dir, f{cache_key}.pkl) # 检查缓存是否存在且未过期 if os.path.exists(cache_file): file_mtime os.path.getmtime(cache_file) if time.time() - file_mtime self.ttl: with open(cache_file, rb) as f: return pickle.load(f) # 执行查询 result pywencai.get(queryquery, cookieself.cookie, **kwargs) # 保存到缓存 if result is not None: with open(cache_file, wb) as f: pickle.dump(result, f) return result故障排查指南常见问题及解决方案问题现象可能原因解决方案403 Forbidden错误Cookie过期或无效重新获取最新Cookie检查Cookie格式连接超时网络问题或接口繁忙增加retry参数设置代理服务器数据格式异常接口返回结构变化更新pywencai到最新版本Node.js执行错误Node.js未安装或版本过低安装Node.js v16检查环境变量内存占用过高大数据量查询未分页使用loopFalse或设置合适的perpage参数调试模式启用# 启用详细日志 import logging logging.basicConfig(levellogging.DEBUG) # 使用pywencai时启用日志 df pywencai.get( query测试查询, cookieyour_cookie, logTrue, # 启用内部日志 retry5, sleep1 ) # 检查请求详情 print(f请求参数: {df._request_params if hasattr(df, _request_params) else N/A})监控与告警系统import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart class MonitoringSystem: def __init__(self, cookie, alert_emailNone): self.cookie cookie self.alert_email alert_email self.error_count 0 self.max_errors 5 def health_check(self): 系统健康检查 try: # 执行简单查询测试 test_result pywencai.get( query上证指数, cookieself.cookie, loopFalse, perpage1, logFalse ) if test_result is None or test_result.empty: self.error_count 1 return False, 查询返回空结果 else: self.error_count 0 return True, 系统正常 except Exception as e: self.error_count 1 return False, str(e) def send_alert(self, subject, message): 发送告警邮件 if not self.alert_email: return msg MIMEMultipart() msg[From] monitoryourdomain.com msg[To] self.alert_email msg[Subject] subject msg.attach(MIMEText(message, plain)) try: with smtplib.SMTP(smtp.yourdomain.com, 587) as server: server.starttls() server.login(username, password) server.send_message(msg) except Exception as e: print(f发送告警失败: {e}) def start_monitoring(self, interval300): 启动监控循环 import time import schedule def check_and_alert(): healthy, message self.health_check() if not healthy: print(f系统异常: {message}) if self.error_count self.max_errors: alert_msg f pywencai系统告警 时间: {datetime.now()} 错误计数: {self.error_count} 错误信息: {message} 建议: 检查Cookie有效性或网络连接 self.send_alert(pywencai系统告警, alert_msg) return healthy # 每5分钟检查一次 schedule.every(interval).seconds.do(check_and_alert) while True: schedule.run_pending() time.sleep(1) # 启动监控 monitor MonitoringSystem( cookieyour_cookie, alert_emailadminyourdomain.com ) # 在后台线程中运行监控 monitor_thread threading.Thread(targetmonitor.start_monitoring) monitor_thread.daemon True monitor_thread.start()扩展开发与最佳实践自定义数据处理器from pywencai import get import pandas as pd class EnhancedWencai: def __init__(self, cookie, custom_handlersNone): self.cookie cookie self.custom_handlers custom_handlers or {} def get_enhanced(self, query, **kwargs): 增强版数据获取支持自定义处理 # 获取原始数据 raw_data get(queryquery, cookieself.cookie, **kwargs) if raw_data is None: return None # 应用自定义处理器 if isinstance(raw_data, pd.DataFrame): return self._process_dataframe(raw_data, query) elif isinstance(raw_data, dict): return self._process_dict(raw_data, query) else: return raw_data def _process_dataframe(self, df, query): 处理DataFrame类型数据 # 添加元数据 df.attrs[query] query df.attrs[fetch_time] datetime.now() # 应用自定义列处理 for col in df.columns: if col in self.custom_handlers: df[col] df[col].apply(self.custom_handlers[col]) # 标准化列名 df.columns [self._standardize_col_name(col) for col in df.columns] return df def _standardize_col_name(self, col_name): 标准化列名 # 移除特殊字符转换为小写等 return col_name.replace( , _).replace((, ).replace(), ).lower() def _process_dict(self, data_dict, query): 处理字典类型数据 processed {} for key, value in data_dict.items(): if isinstance(value, pd.DataFrame): processed[key] self._process_dataframe(value, query) else: processed[key] value return processed # 使用自定义处理器 custom_handlers { 市盈率: lambda x: float(x) if x ! - else None, 总市值: lambda x: float(str(x).replace(亿, )) * 1e8 if 亿 in str(x) else float(x) } enhanced EnhancedWencai( cookieyour_cookie, custom_handlerscustom_handlers ) # 获取并处理数据 processed_data enhanced.get_enhanced( query沪深300成分股, loopTrue, sort_key总市值, sort_orderdesc )版本兼容性说明pywencai当前版本为0.13.1支持Python 3.8版本。主要依赖包括requests- HTTP请求库pandas- 数据处理和分析pydash- 数据操作工具fake-useragent- 随机用户代理生成PyExecJS- JavaScript执行环境使用限制与注意事项Cookie有效期同花顺问财Cookie通常有有效期限制需要定期更新请求频率避免高频请求建议单次请求间隔至少1秒数据用途仅限学习和研究使用商业用途需评估法律风险接口稳定性问财接口可能随时变更需关注项目更新错误处理建议实现完整的错误处理和重试机制总结与展望pywencai作为一个高效的同花顺问财数据获取工具通过简洁的API接口和智能的数据处理机制为金融数据分析和量化研究提供了强大支持。其模块化架构、完善的错误处理机制和灵活的扩展性使其成为金融数据分析领域的理想选择。通过合理的配置和最佳实践开发者可以构建稳定可靠的金融数据管道支持从简单的数据查询到复杂的实时监控系统等各种应用场景。随着金融科技的发展pywencai将继续演进为开发者提供更加强大和易用的数据获取能力。图2数据与交易知识星球社群二维码提供更多金融数据工具资源和技术交流立即开始你的金融数据之旅克隆项目仓库git clone https://gitcode.com/gh_mirrors/py/pywencai安装依赖pip install pywencai获取Cookie凭证并开始你的第一个查询探索更多高级功能和集成方案通过合理运用pywencai你将能够快速构建个性化的金融数据分析系统大幅提升数据处理效率让Python金融数据分析变得更加简单高效。【免费下载链接】pywencai获取同花顺问财数据项目地址: https://gitcode.com/gh_mirrors/py/pywencai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
Python金融数据获取终极指南：使用pywencai高效访问同花顺问财数据
发布时间：2026/5/17 2:14:26
Python金融数据获取终极指南使用pywencai高效访问同花顺问财数据【免费下载链接】pywencai获取同花顺问财数据项目地址: https://gitcode.com/gh_mirrors/py/pywencai在金融数据分析和量化投资领域获取高质量、实时的市场数据是每个数据分析师和开发者面临的核心挑战。传统方法如网页爬虫开发复杂、维护成本高而官方API往往价格昂贵或功能有限。pywencai作为一个开源Python工具为开发者提供了高效、稳定的同花顺问财数据访问方案让金融数据获取变得简单可靠。项目价值主张为什么选择pywencaipywencai的核心价值在于其智能化的一站式金融数据获取能力。通过模拟浏览器行为并解析同花顺问财接口该工具将复杂的网页数据提取过程封装为简洁的Python API让开发者能够专注于数据分析而非数据获取的技术细节。核心优势对比特性维度pywencai解决方案传统网页爬虫官方API接口开发复杂度极低几行代码即可使用高需要处理反爬、解析等中等需要学习API文档数据完整性完整支持所有问财功能有限依赖网页结构完整但可能有功能限制稳定性保障内置重试机制和错误处理脆弱易受网站变更影响稳定但可能有调用限制成本效益完全免费开源免费但维护成本高通常需要付费实时性接近实时与网页同步实时但易被封禁实时但有速率限制扩展性易于集成到现有系统集成复杂标准API集成技术架构创新pywencai采用模块化设计核心架构包含三个关键组件请求引擎(wencai.py) - 处理HTTP通信和重试逻辑数据转换器(convert.py) - 解析JSON响应为DataFrame请求头生成器(headers.py) - 动态生成合法请求头这种分层架构确保了代码的可维护性和扩展性每个模块都专注于单一职责。核心功能演示5分钟快速上手环境配置与安装首先确保系统已安装Node.js v16这是执行JavaScript代码生成请求令牌的前提# 安装pywencai pip install pywencai # 验证安装 python -c import pywencai; print(安装成功)获取Cookie凭证访问同花顺问财网站获取Cookie是使用pywencai的关键步骤使用Chrome浏览器访问www.iwencai.com按F12打开开发者工具切换到Network标签页刷新页面并选择任意POST请求在请求头中找到Cookie字段并复制完整值图1浏览器开发者工具中获取Cookie的详细步骤红色箭头标注关键Cookie字段位置基础数据查询示例import pywencai import pandas as pd # 基础查询获取沪深300成分股中市盈率低于30的股票 df pywencai.get( query沪深300成分股 市盈率30, cookieyour_cookie_value_here, # 替换为实际Cookie loopTrue, # 自动分页获取全部数据 perpage100, # 每页数据量 sort_key市盈率, # 排序字段 sort_orderasc # 升序排列 ) print(f获取到 {len(df)} 条数据) print(df[[股票代码, 股票名称, 市盈率, 总市值]].head()) # 保存到CSV文件 df.to_csv(low_pe_stocks.csv, indexFalse, encodingutf-8-sig)高级查询功能# 多条件组合查询 complex_query pywencai.get( query连续3年ROE15% 资产负债率50% 市值200亿, cookieyour_cookie_value, query_typestock, loopTrue, logTrue # 启用日志输出 ) # 特定股票查询 specific_stocks pywencai.get( query股票, cookieyour_cookie_value, find[600519, 000858, 000001], # 贵州茅台、五粮液、平安银行 perpage50 )架构设计解析深入理解技术实现请求流程架构核心模块详解1. 请求头生成机制 (headers.py)# headers.py核心代码片段 def get_token(): 获取hexin-v令牌 result subprocess.run( [node, os.path.join(os.path.dirname(__file__), hexin-v.bundle.js)], stdoutsubprocess.PIPE ) return result.stdout.decode().strip() def headers(cookieNone, user_agentNone): 构建请求头 if user_agent is None: from fake_useragent import UserAgent ua UserAgent() user_agent ua.random return { hexin-v: get_token(), # 动态生成的加密令牌 User-Agent: user_agent, # 随机用户代理 cookie: cookie # 用户提供的Cookie }该模块通过执行JavaScript代码生成hexin-v令牌这是同花顺问财接口的重要验证参数。使用Node.js执行预编译的JavaScript文件确保令牌生成的准确性和实时性。2. 智能重试机制 (wencai.py)def while_do(do, retry10, sleep0, logFalse): 智能重试装饰器 count 0 while count retry: time.sleep(sleep) try: return do() except: log and logger.warning(f{count1}次尝试失败) count 1 return None重试机制支持指数退避策略在网络不稳定或接口临时限制时自动重试默认最多10次确保数据获取的可靠性。3. 数据转换引擎 (convert.py)def convert(res): 处理API响应数据 result json.loads(res.text) content _.get(result, data.answer.0.txt.0.content) if type(content) str: content json.loads(content) components content[components] # 根据组件类型选择不同处理器 if (len(components) 1 and _.get(components[0], show_type) xuangu_tableV1): # 处理选股表格数据 return xuangu_tableV1_handler(components[0], components) else: # 处理复杂数据结构 return multi_show_type_handler(components)数据转换器支持多种数据类型包括表格数据、文本数据、嵌套结构等确保返回统一的数据格式。配置文件说明项目的配置文件结构清晰便于维护pyproject.toml- Python项目配置和依赖管理package.json- Node.js依赖管理包含jsdom等JavaScript运行时依赖webpack.config.js- JavaScript打包配置用于生成hexin-v.bundle.js实战应用场景金融数据分析案例场景1价值投资筛选系统class ValueInvestmentScreener: def __init__(self, cookie): self.cookie cookie def screen_by_factors(self, factors_config): 基于多因子进行股票筛选 factors_config: 因子配置字典 results {} for factor_name, query in factors_config.items(): try: df pywencai.get( queryquery, cookieself.cookie, loopTrue, logFalse, retry5, sleep0.5 # 避免请求过于频繁 ) if not df.empty: # 数据清洗和验证 df self._clean_data(df) results[factor_name] df print(f{factor_name}因子筛选完成: {len(df)}只股票) else: print(f{factor_name}因子未找到匹配股票) except Exception as e: print(f{factor_name}因子筛选失败: {str(e)}) continue return results def _clean_data(self, df): 数据清洗和标准化 # 去除重复数据 df df.drop_duplicates(subset[股票代码]) # 处理缺失值 numeric_columns [市盈率, 市净率, ROE, 净利润增长率] for col in numeric_columns: if col in df.columns: df[col] pd.to_numeric(df[col], errorscoerce) return df # 使用示例 screener ValueInvestmentScreener(cookieyour_cookie) factors { 高ROE低负债: 连续3年ROE15% 资产负债率50%, 低估值高成长: 市盈率20 净利润增长率30%, 高股息率: 股息率3% 连续3年分红 } results screener.screen_by_factors(factors)场景2技术指标监控系统import schedule import time from datetime import datetime class TechnicalIndicatorMonitor: def __init__(self, cookie, indicators): self.cookie cookie self.indicators indicators self.history_data {} def monitor_market_signals(self): 监控市场技术信号 signals {} for indicator_name, query in self.indicators.items(): try: df pywencai.get( queryquery, cookieself.cookie, loopFalse, # 实时监控不需要分页 perpage20, # 只获取前20条 sort_key涨幅, sort_orderdesc ) if not df.empty: signals[indicator_name] { count: len(df), top_stocks: df[[股票代码, 股票名称, 涨幅]].head(5).to_dict(records), timestamp: datetime.now().strftime(%Y-%m-%d %H:%M:%S) } except Exception as e: print(f指标{indicator_name}监控失败: {str(e)}) return signals def start_daily_monitoring(self): 启动每日监控任务 schedule.every().day.at(09:30).do(self.daily_morning_report) schedule.every().day.at(15:00).do(self.daily_closing_report) while True: schedule.run_pending() time.sleep(60) # 技术指标配置 indicators { MACD金叉: MACD金叉 成交量放大, 突破均线: 股价站上20日均线 成交量5日均量, 强势反弹: 涨幅5% 换手率3% } monitor TechnicalIndicatorMonitor(cookieyour_cookie, indicatorsindicators)场景3行业数据分析报告import pandas as pd import matplotlib.pyplot as plt class IndustryAnalysis: def __init__(self, cookie): self.cookie cookie def analyze_industry_trend(self, industries, metrics): 分析多个行业的数据趋势 industries: 行业列表 metrics: 分析指标列表 industry_data {} for industry in industries: print(f正在分析{industry}行业...) # 构建行业查询 query f{industry}行业 {metrics} try: df pywencai.get( queryquery, cookieself.cookie, loopTrue, query_typestock, sleep1 # 行业间请求间隔 ) if not df.empty: # 数据预处理 df self._preprocess_industry_data(df, metrics) industry_data[industry] df # 生成行业统计 stats self._calculate_industry_stats(df, metrics) print(f{industry}行业分析完成: {stats}) except Exception as e: print(f{industry}行业分析失败: {str(e)}) continue return industry_data def generate_comparison_report(self, industry_data): 生成行业对比报告 comparison_df pd.DataFrame() for industry, df in industry_data.items(): if not df.empty: # 计算行业平均指标 avg_values {} numeric_cols df.select_dtypes(include[float64, int64]).columns for col in numeric_cols: if col in df.columns: avg_values[f{industry}_{col}_avg] df[col].mean() comparison_df pd.concat([comparison_df, pd.DataFrame([avg_values])], ignore_indexTrue) return comparison_df # 使用示例 analyzer IndustryAnalysis(cookieyour_cookie) industries [新能源, 人工智能, 生物医药, 半导体] metrics 市盈率 市净率 ROE 营业收入增长率 industry_data analyzer.analyze_industry_trend(industries, metrics) report analyzer.generate_comparison_report(industry_data)生态整合方案与企业级系统对接与数据库系统集成import sqlalchemy as sa from sqlalchemy import create_engine, Table, Column, Integer, String, Float, DateTime from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker Base declarative_base() class StockData(Base): __tablename__ stock_data id Column(Integer, primary_keyTrue) stock_code Column(String(10), indexTrue) stock_name Column(String(50)) pe_ratio Column(Float) pb_ratio Column(Float) roe Column(Float) market_cap Column(Float) query_date Column(DateTime) created_at Column(DateTime, defaultdatetime.now) class DataPipeline: def __init__(self, cookie, db_url): self.cookie cookie self.engine create_engine(db_url) Base.metadata.create_all(self.engine) self.Session sessionmaker(bindself.engine) def daily_data_collection(self, queries): 每日数据收集管道 session self.Session() for query_name, query_str in queries.items(): try: df pywencai.get( queryquery_str, cookieself.cookie, loopTrue, sleep0.5 ) if not df.empty: # 数据转换和存储 self._save_to_database(session, df, query_name) print(f{query_name}数据收集完成: {len(df)}条记录) except Exception as e: print(f{query_name}数据收集失败: {str(e)}) session.rollback() continue session.commit() session.close() def _save_to_database(self, session, df, query_name): 保存数据到数据库 for _, row in df.iterrows(): stock_data StockData( stock_coderow.get(股票代码, ), stock_namerow.get(股票名称, ), pe_ratioself._safe_float(row.get(市盈率)), pb_ratioself._safe_float(row.get(市净率)), roeself._safe_float(row.get(ROE)), market_capself._safe_float(row.get(总市值)), query_datedatetime.now() ) session.add(stock_data) def _safe_float(self, value): 安全转换为浮点数 try: return float(value) if value else None except (ValueError, TypeError): return None # 配置数据库连接 pipeline DataPipeline( cookieyour_cookie, db_urlpostgresql://user:passwordlocalhost:5432/finance_db ) # 定义数据收集任务 daily_queries { 沪深300成分股: 沪深300成分股, 高增长股票: 净利润增长率30% 市值100亿, 低估值股票: 市盈率15 市净率2 } pipeline.daily_data_collection(daily_queries)与消息队列系统集成import redis import json from datetime import datetime class RealTimeDataStream: def __init__(self, cookie, redis_hostlocalhost, redis_port6379): self.cookie cookie self.redis_client redis.Redis(hostredis_host, portredis_port, db0) def stream_market_data(self, queries, interval60): 实时数据流推送 import time while True: for stream_name, query in queries.items(): try: # 获取实时数据 df pywencai.get( queryquery, cookieself.cookie, loopFalse, perpage50, sort_key更新时间, sort_orderdesc ) if not df.empty: # 转换为JSON格式 data { stream: stream_name, timestamp: datetime.now().isoformat(), count: len(df), data: df.head(10).to_dict(records) } # 发布到Redis Stream self.redis_client.xadd( market_data_stream, {data: json.dumps(data, ensure_asciiFalse)} ) print(f{stream_name}数据流更新: {len(df)}条记录) except Exception as e: print(f{stream_name}数据流更新失败: {str(e)}) # 等待下一个周期 time.sleep(interval) def get_latest_data(self, stream_name, count10): 获取最新的数据 messages self.redis_client.xrevrange( market_data_stream, , -, countcount ) return [ json.loads(msg[1][bdata].decode(utf-8)) for msg in messages if json.loads(msg[1][bdata].decode(utf-8))[stream] stream_name ] # 实时数据流配置 streamer RealTimeDataStream(cookieyour_cookie) stream_queries { hot_stocks: 涨幅5% 换手率3%, volume_leaders: 成交量排名前50, breakout_stocks: 创60日新高 } # 在后台线程中启动数据流 import threading stream_thread threading.Thread( targetstreamer.stream_market_data, args(stream_queries, 60) # 60秒间隔 ) stream_thread.daemon True stream_thread.start()性能优化与故障排查性能优化技巧批量请求优化# 使用连接池和会话重用 import requests from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry session requests.Session() retry_strategy Retry( total3, backoff_factor1, status_forcelist[429, 500, 502, 503, 504] ) adapter HTTPAdapter(max_retriesretry_strategy) session.mount(http://, adapter) session.mount(https://, adapter) # 在pywencai中使用自定义会话 df pywencai.get( query沪深300, cookieyour_cookie, request_params{session: session} )缓存机制实现import hashlib import pickle from functools import lru_cache import os class CachedWencai: def __init__(self, cookie, cache_dir./cache, ttl3600): self.cookie cookie self.cache_dir cache_dir self.ttl ttl # 缓存有效期秒 if not os.path.exists(cache_dir): os.makedirs(cache_dir) def get_with_cache(self, query, **kwargs): 带缓存的查询 # 生成缓存键 cache_key hashlib.md5( f{query}_{str(kwargs)}.encode() ).hexdigest() cache_file os.path.join(self.cache_dir, f{cache_key}.pkl) # 检查缓存是否存在且未过期 if os.path.exists(cache_file): file_mtime os.path.getmtime(cache_file) if time.time() - file_mtime self.ttl: with open(cache_file, rb) as f: return pickle.load(f) # 执行查询 result pywencai.get(queryquery, cookieself.cookie, **kwargs) # 保存到缓存 if result is not None: with open(cache_file, wb) as f: pickle.dump(result, f) return result故障排查指南常见问题及解决方案问题现象可能原因解决方案403 Forbidden错误Cookie过期或无效重新获取最新Cookie检查Cookie格式连接超时网络问题或接口繁忙增加retry参数设置代理服务器数据格式异常接口返回结构变化更新pywencai到最新版本Node.js执行错误Node.js未安装或版本过低安装Node.js v16检查环境变量内存占用过高大数据量查询未分页使用loopFalse或设置合适的perpage参数调试模式启用# 启用详细日志 import logging logging.basicConfig(levellogging.DEBUG) # 使用pywencai时启用日志 df pywencai.get( query测试查询, cookieyour_cookie, logTrue, # 启用内部日志 retry5, sleep1 ) # 检查请求详情 print(f请求参数: {df._request_params if hasattr(df, _request_params) else N/A})监控与告警系统import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart class MonitoringSystem: def __init__(self, cookie, alert_emailNone): self.cookie cookie self.alert_email alert_email self.error_count 0 self.max_errors 5 def health_check(self): 系统健康检查 try: # 执行简单查询测试 test_result pywencai.get( query上证指数, cookieself.cookie, loopFalse, perpage1, logFalse ) if test_result is None or test_result.empty: self.error_count 1 return False, 查询返回空结果 else: self.error_count 0 return True, 系统正常 except Exception as e: self.error_count 1 return False, str(e) def send_alert(self, subject, message): 发送告警邮件 if not self.alert_email: return msg MIMEMultipart() msg[From] monitoryourdomain.com msg[To] self.alert_email msg[Subject] subject msg.attach(MIMEText(message, plain)) try: with smtplib.SMTP(smtp.yourdomain.com, 587) as server: server.starttls() server.login(username, password) server.send_message(msg) except Exception as e: print(f发送告警失败: {e}) def start_monitoring(self, interval300): 启动监控循环 import time import schedule def check_and_alert(): healthy, message self.health_check() if not healthy: print(f系统异常: {message}) if self.error_count self.max_errors: alert_msg f pywencai系统告警 时间: {datetime.now()} 错误计数: {self.error_count} 错误信息: {message} 建议: 检查Cookie有效性或网络连接 self.send_alert(pywencai系统告警, alert_msg) return healthy # 每5分钟检查一次 schedule.every(interval).seconds.do(check_and_alert) while True: schedule.run_pending() time.sleep(1) # 启动监控 monitor MonitoringSystem( cookieyour_cookie, alert_emailadminyourdomain.com ) # 在后台线程中运行监控 monitor_thread threading.Thread(targetmonitor.start_monitoring) monitor_thread.daemon True monitor_thread.start()扩展开发与最佳实践自定义数据处理器from pywencai import get import pandas as pd class EnhancedWencai: def __init__(self, cookie, custom_handlersNone): self.cookie cookie self.custom_handlers custom_handlers or {} def get_enhanced(self, query, **kwargs): 增强版数据获取支持自定义处理 # 获取原始数据 raw_data get(queryquery, cookieself.cookie, **kwargs) if raw_data is None: return None # 应用自定义处理器 if isinstance(raw_data, pd.DataFrame): return self._process_dataframe(raw_data, query) elif isinstance(raw_data, dict): return self._process_dict(raw_data, query) else: return raw_data def _process_dataframe(self, df, query): 处理DataFrame类型数据 # 添加元数据 df.attrs[query] query df.attrs[fetch_time] datetime.now() # 应用自定义列处理 for col in df.columns: if col in self.custom_handlers: df[col] df[col].apply(self.custom_handlers[col]) # 标准化列名 df.columns [self._standardize_col_name(col) for col in df.columns] return df def _standardize_col_name(self, col_name): 标准化列名 # 移除特殊字符转换为小写等 return col_name.replace( , _).replace((, ).replace(), ).lower() def _process_dict(self, data_dict, query): 处理字典类型数据 processed {} for key, value in data_dict.items(): if isinstance(value, pd.DataFrame): processed[key] self._process_dataframe(value, query) else: processed[key] value return processed # 使用自定义处理器 custom_handlers { 市盈率: lambda x: float(x) if x ! - else None, 总市值: lambda x: float(str(x).replace(亿, )) * 1e8 if 亿 in str(x) else float(x) } enhanced EnhancedWencai( cookieyour_cookie, custom_handlerscustom_handlers ) # 获取并处理数据 processed_data enhanced.get_enhanced( query沪深300成分股, loopTrue, sort_key总市值, sort_orderdesc )版本兼容性说明pywencai当前版本为0.13.1支持Python 3.8版本。主要依赖包括requests- HTTP请求库pandas- 数据处理和分析pydash- 数据操作工具fake-useragent- 随机用户代理生成PyExecJS- JavaScript执行环境使用限制与注意事项Cookie有效期同花顺问财Cookie通常有有效期限制需要定期更新请求频率避免高频请求建议单次请求间隔至少1秒数据用途仅限学习和研究使用商业用途需评估法律风险接口稳定性问财接口可能随时变更需关注项目更新错误处理建议实现完整的错误处理和重试机制总结与展望pywencai作为一个高效的同花顺问财数据获取工具通过简洁的API接口和智能的数据处理机制为金融数据分析和量化研究提供了强大支持。其模块化架构、完善的错误处理机制和灵活的扩展性使其成为金融数据分析领域的理想选择。通过合理的配置和最佳实践开发者可以构建稳定可靠的金融数据管道支持从简单的数据查询到复杂的实时监控系统等各种应用场景。随着金融科技的发展pywencai将继续演进为开发者提供更加强大和易用的数据获取能力。图2数据与交易知识星球社群二维码提供更多金融数据工具资源和技术交流立即开始你的金融数据之旅克隆项目仓库git clone https://gitcode.com/gh_mirrors/py/pywencai安装依赖pip install pywencai获取Cookie凭证并开始你的第一个查询探索更多高级功能和集成方案通过合理运用pywencai你将能够快速构建个性化的金融数据分析系统大幅提升数据处理效率让Python金融数据分析变得更加简单高效。【免费下载链接】pywencai获取同花顺问财数据项目地址: https://gitcode.com/gh_mirrors/py/pywencai创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
