import os
import win32com.client #pip install pywin32 -i https://mirrors.aliyun.com/pypi/simple/
import sys
import datetime
def get_real_path():
"""兼容开发与打包环境的路径获取"""
if getattr(sys, 'frozen', False):
base_dir = os.path.dirname(sys.executable)
else:
base_dir = os.path.dirname(os.path.abspath(__file__))
return base_dir
def gen_output_folder(folder):
timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
output_folder = os.path.join(folder, f"pdf_{timestamp}")
os.makedirs(output_folder, exist_ok=True)
return output_folder
def get_word_files_from_current_folder(folder):
word_files = []
for file in os.listdir(folder):
if file.endswith(".doc") or file.endswith(".docx"):
word_files.append(os.path.join(folder, file))
return word_files
def detect_office_or_wps():
try:
word = win32com.client.gencache.EnsureDispatch("Word.Application")
return "office"
except:
try:
wps = win32com.client.gencache.EnsureDispatch("Kwps.Application")
return "wps"
except:
return None
def convert_word_to_pdf_auto(input_path, output_path, engine):
if engine == "office":
app = win32com.client.Dispatch("Word.Application")
elif engine == "wps":
app = win32com.client.Dispatch("Kwps.Application")
else:
print("没有检测到可用的 Office 或 WPS")
return
app.Visible = False
try:
doc = app.Documents.Open(input_path)
doc.SaveAs(output_path, FileFormat=17)
doc.Close()
print(f"转换成功:{input_path}")
except Exception as e:
print(f"转换失败:{input_path},原因:{e}")
try:
app.Quit()
except:
print("当前环境不支持 Quit,跳过退出。")
def batch_convert_here():
engine = detect_office_or_wps()
if not engine:
print("系统里没有安装 Office 或 WPS,没法转换")
return
folder = get_real_path()
word_files = get_word_files_from_current_folder(folder)
if not word_files:
print("当前文件夹没有发现 Word 文件")
return
output_folder = gen_output_folder(folder)
for word_file in word_files:
filename = os.path.splitext(os.path.basename(word_file))[0]
pdf_path = os.path.join(output_folder, f"{filename}.pdf")
convert_word_to_pdf_auto(word_file, pdf_path, engine)
print(f"所有文件转换完成啦!PDF 都在 {output_folder} 文件夹里")
if __name__ == "__main__":
try:
batch_convert_here()
print("按 Enter 键退出...")
input()
except Exception as e:
print(e)
print("程序运行错误,按 Enter 键退出...")
input()
分类: Python
-
指定目录下word文档批量转PDF
-
python在windows电脑找回WiFi密码
不是破解密码,在能无线上网的情况子下,找出保存过的密码
import subprocess # 执行netsh命令获取Wi-Fi密码 result = subprocess.check_output('netsh wlan show profiles', shell=True) # Convert bytes to string using the default encoding result = result.decode() print(result) profiles = [] for line in result.split('\n'): if "所有用户配置文件 :" in line: profiles.append(line.split(':')[1].strip()) for profile in profiles: try: password_result = subprocess.check_output(f'netsh wlan show profile name="{profile}" key=clear', shell=True) # Convert bytes to string using the default encoding password_result = password_result.decode() # print(password_result) for line in password_result.split('\n'): if "关键内容" in line: print('{:<40}{:>30}'.format('WIFI名称:'+profile,'WIFI密码:'+line.split(':')[1].strip())) except: print(f'无法获取{profile}的密码')更多解释的代码:
import subprocess
import re
import ctypes
import sys
def is_admin():
try:
return ctypes.windll.shell32.IsUserAnAdmin()
except:
return False
def get_current_wifi_name():
try:
# 使用系统默认编码解码
output = subprocess.check_output(‘netsh wlan show interface’, shell=True).decode(sys.getdefaultencoding(), errors=’replace’)
# 查找当前连接的Wi-Fi名称
match = re.search(r”SSID\s*:\s*(.*)”, output)
if match:
return match.group(1).strip()
else:
print(“无法获取当前连接的Wi-Fi名称”)
return None
except Exception as e:
print(f”发生错误: {e}”)
return None
def get_wifi_password(ssid):
if not ssid:
return None
try:
# 使用系统默认编码解码
output = subprocess.check_output(f’netsh wlan show profile name=”{ssid}” key=clear’, shell=True).decode(sys.getdefaultencoding(), errors=’replace’)
print(“netsh 命令完整输出:”)
print(output) # 打印完整输出
# 查找Wi-Fi密码,使用更灵活的正则表达式
match = re.search(r”关键内容\s*:\s*(.*)”, output)
if match:
return match.group(1).strip()
else:
print(f”未找到Wi-Fi ‘{ssid}’的密码,可能该网络没有保存密码或为开放网络。”)
return None
except subprocess.CalledProcessError as e:
print(f”执行命令时出错: {e.output.decode(sys.getdefaultencoding(), errors=’replace’)}”)
return None
except Exception as e:
print(f”发生错误: {e}”)
return None
if __name__ == “__main__”:
if not is_admin():
ctypes.windll.shell32.ShellExecuteW(None, “runas”, sys.executable, ” “.join(sys.argv), None, 1)
else:
current_wifi = get_current_wifi_name()
if current_wifi:
print(f”当前连接的Wi-Fi: {current_wifi}”)
password = get_wifi_password(current_wifi)
if password:
print(f”Wi-Fi密码: {password}”)
else:
print(“未能获取密码。请确保该网络保存了密码。”)
else:
print(“无法确定当前连接的Wi-Fi。”)
# 等待用户输入后再关闭窗口
input(“按回车键退出…”)有详细解释的代码:
import subprocess import re import ctypes import sys def is_admin(): """ 检查当前 Python 脚本是否以管理员权限运行。 返回: bool: 如果以管理员权限运行返回 True,否则返回 False。 """ try: return ctypes.windll.shell32.IsUserAnAdmin() except (AttributeError, OSError): # 处理访问动态链接库函数失败或系统错误 return False def get_current_wifi_name(): # 定义一个名为 get_current_wifi_name 的函数,该函数不接受任何参数。 try: # 使用 try 语句块来捕获可能出现的异常。 # 使用系统默认编码解码 output = subprocess.check_output('netsh wlan show interface', shell=True).decode(sys.getdefaultencoding(), errors='replace') # subprocess.check_output 用于在 Python 中执行外部命令。 # 'netsh wlan show interface' 是 Windows 系统的命令,用于显示当前 Wi-Fi 接口的信息。 # shell=True 表示通过系统的 shell 来执行命令。 # decode(sys.getdefaultencoding(), errors='replace') 将命令执行结果的字节流按系统默认编码解码为字符串, # errors='replace' 表示遇到无法解码的字符时用替换字符替代。 # 查找当前连接的Wi-Fi名称 match = re.search(r"SSID\s*:\s*(.*)", output) # re.search 是 Python 中 re 模块的函数,用于在字符串中搜索匹配正则表达式的第一个位置。 # r"SSID\s*:\s*(.*)" 是正则表达式,含义如下: # "SSID" 匹配字面字符串 "SSID"; # \s* 匹配零个或多个空白字符; # ":" 匹配冒号; # \s* 再次匹配零个或多个空白字符; # (.*) 是一个捕获组,匹配任意字符(除换行符外)零次或多次,用于捕获 Wi-Fi 名称。 if match: return match.group(1).strip() # 如果匹配成功,match.group(1) 获取捕获组中的内容,即 Wi-Fi 名称。 # strip() 方法去除字符串两端的空白字符后返回。 else: print("无法获取当前连接的Wi-Fi名称") return None # 如果匹配失败,打印提示信息并返回 None。 except Exception as e: print(f"发生错误: {e}") return None # 捕获所有异常,打印错误信息并返回 None。 def get_wifi_password(ssid): # 定义一个名为 get_wifi_password 的函数,接收一个参数 ssid,代表 Wi-Fi 网络的名称。 if not ssid: # 检查传入的 ssid 是否为空(即 None、空字符串等假值)。 return None # 如果 ssid 为空,直接返回 None。 try: # 使用 try 语句块来捕获可能出现的异常。 # 使用系统默认编码解码 output = subprocess.check_output(f'netsh wlan show profile name="{ssid}" key=clear', shell=True).decode(sys.getdefaultencoding(), errors='replace') # subprocess.check_output 用于在 Python 中执行外部命令。 # f'netsh wlan show profile name="{ssid}" key=clear' 是 Windows 系统的命令,用于显示指定 Wi-Fi 网络的详细配置信息,包括密码(key=clear 表示显示明文密码)。 # shell=True 表示通过系统的 shell 来执行命令。 # decode(sys.getdefaultencoding(), errors='replace') 将命令执行结果的字节流按系统默认编码解码为字符串, # errors='replace' 表示遇到无法解码的字符时用替换字符替代。 print("netsh 命令完整输出:") print(output) # 打印完整输出 # 查找Wi-Fi密码,使用更灵活的正则表达式 match = re.search(r"关键内容\s*:\s*(.*)", output) # re.search 是 Python 中 re 模块的函数,用于在字符串中搜索匹配正则表达式的第一个位置。 # r"关键内容\s*:\s*(.*)" 是正则表达式,含义如下: # "关键内容" 匹配中文的 "关键内容" 字符串; # \s* 匹配零个或多个空白字符; # ":" 匹配冒号; # \s* 再次匹配零个或多个空白字符; # (.*) 是一个捕获组,匹配任意字符(除换行符外)零次或多次,用于捕获 Wi-Fi 密码。 if match: return match.group(1).strip() # 如果匹配成功,match.group(1) 获取捕获组中的内容,即 Wi-Fi 密码。 # strip() 方法去除字符串两端的空白字符后返回。 else: print(f"未找到Wi-Fi '{ssid}'的密码,可能该网络没有保存密码或为开放网络。") return None # 如果匹配失败,打印提示信息并返回 None。 except subprocess.CalledProcessError as e: print(f"执行命令时出错: {e.output.decode(sys.getdefaultencoding(), errors='replace')}") return None # 捕获 subprocess.check_output 执行命令失败时抛出的 CalledProcessError 异常, # 打印错误信息并返回 None。 except Exception as e: print(f"发生错误: {e}") return None # 捕获其他所有异常,打印错误信息并返回 None。 if __name__ == "__main__": if not is_admin(): ctypes.windll.shell32.ShellExecuteW(None, "runas", sys.executable, " ".join(sys.argv), None, 1) else: current_wifi = get_current_wifi_name() if current_wifi: print(f"当前连接的Wi-Fi: {current_wifi}") password = get_wifi_password(current_wifi) if password: print(f"Wi-Fi密码: {password}") else: print("未能获取密码。请确保该网络保存了密码。") else: print("无法确定当前连接的Wi-Fi。") # 等待用户输入后再关闭窗口 input("按回车键退出...") -
python waitress模块 可替代flask 的开发环境吗
是的,Python的
waitress模块可以替代Flask自带的开发服务器,而且在生产环境中更推荐使用waitress这样的WSGI服务器。以下是详细说明:1. Flask自带服务器的局限性
Flask内置的服务器(通过
app.run()启动)是为开发环境设计的,它:- 单线程运行:无法处理并发请求,性能差。
- 安全性低:缺乏生产级的安全加固。
- 稳定性不足:长时间运行可能崩溃,不适合高负载场景。
2. Waitress的优势
waitress是一个纯Python实现的WSGI服务器,专为生产环境设计,具有以下特点:- 多线程支持:默认启用多线程处理请求,大幅提升并发能力。
- 跨平台兼容:支持Linux、Windows、macOS等。
- 稳定性高:适合长时间运行的应用。
- 安装简单:通过
pip install waitress即可安装。
3. 如何用Waitress运行Flask应用
以下是一个示例,展示如何将Flask应用从开发服务器切换到
waitress:from flask import Flask app = Flask(__name__) @app.route('/') def hello(): return "Hello, World!" if __name__ == '__main__': # 开发环境:使用Flask内置服务器 # app.run(debug=True) # 生产环境:使用waitress服务器 from waitress import serve serve(app, host='0.0.0.0', port=8080)关键区别:
- 开发时:使用
app.run(),可开启调试模式(debug=True)。 - 部署时:使用
serve(app, host='0.0.0.0', port=8080),关闭调试模式。
4. 生产环境的其他建议
- 反向代理:在
waitress前使用Nginx或Apache作为反向代理,处理静态文件和负载均衡。 - 进程管理器:结合
systemd或supervisor管理应用进程,确保自动重启。 - 环境分离:使用
.env文件或环境变量区分开发和生产配置。
总结
waitress完全可以替代Flask的开发服务器,并且更适合生产环境。它提供了更好的性能和稳定性,同时保持了Python的简洁性。 -
获取A股行业、股价信息
""" 股票数据获取程序,并写入get_stocks.xslx文件 """ import pandas as pd import requests import logging import os def fetch_stock_data(stock_code): """ 获取股票数据的核心函数 参数: stock_code (str): 股票代码,如"688360" 返回: dict: 包含股票数据的字典 """ # 配置日志记录 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') logger = logging.getLogger('StockDataFetcher') # 原始API端点 api_url = "https://push2.eastmoney.com/api/qt/stock/get" try: # 6开头为沪市(代码1),其他为深市(代码0) market_code = 1 if stock_code.startswith("6") else 0 # 构建请求参数 request_params = { "fltt": "2", "invt": "2", "fields": "f57,f58,f84,f85,f127,f116,f117,f189,f43", "secid": f"{market_code}.{stock_code}", } logger.info(f"正在请求股票数据: {stock_code}") # 发送API请求(超时处理) response = requests.get(api_url, params=request_params, timeout=10) response.raise_for_status() # 检查HTTP状态码 logger.info("数据获取成功") return response.json() except requests.exceptions.RequestException as e: logger.error(f"请求失败: {e}") return None except Exception as e: logger.error(f"处理过程中发生错误: {e}") return None def process_stock_data(raw_data): """ 处理从API获取的原始股票数据 参数: raw_data (dict): API返回的原始数据 返回: pd.DataFrame: 处理后的股票数据 """ if not raw_data or "data" not in raw_data: return pd.DataFrame() # 原始字段到中文名称的映射 field_mapping = { "f57": "股票代码", "f58": "股票简称", "f84": "总股本", "f85": "流通股", "f127": "行业", "f116": "总市值", "f117": "流通市值", "f189": "上市时间", "f43": "最新价", } try: # 提取核心数据 stock_data = raw_data.get("data", {}) # 创建数据列表 - 使用列表推导式更高效 processed_data = [] for field, display_name in field_mapping.items(): # 只处理存在的字段 if field in stock_data: processed_data.append({ "字段标识": field, "项目": display_name, "数值": stock_data[field] }) # 转换为DataFrame df = pd.DataFrame(processed_data) # 如果数据为空,返回空DataFrame if df.empty: return df # 仅保留需要的列 return df[["项目", "数值"]] except Exception as e: logging.error(f"数据处理失败: {e}") return pd.DataFrame() def display_stock_info(stock_code): """ 显示股票信息的主函数 参数: stock_code (str): 股票代码 """ # 获取原始数据 raw_data = fetch_stock_data(stock_code) if not raw_data: print(f"无法获取股票 {stock_code} 的数据") return # 处理数据 stock_df = process_stock_data(raw_data) if stock_df.empty: print(f"股票 {stock_code} 的数据处理失败") return # 格式化输出 print("\n股票信息:") print(stock_df.to_string(index=False)) print("\n") file_path = 'get_stocks.xlsx' try: # 检查文件是否存在 file_exists = os.path.isfile(file_path) mode = 'a' if file_exists else 'w' writer_kwargs = { 'engine': 'openpyxl', 'mode': mode } if mode == 'a': writer_kwargs['if_sheet_exists'] = 'replace' with pd.ExcelWriter(file_path, **writer_kwargs) as writer: stock_df.to_excel(writer, sheet_name=stock_code, index=False) print(f"股票 {stock_code} 的数据已成功写入 {file_path}") except Exception as e: print(f"写入股票 {stock_code} 数据到 Excel 时出错: {e}") # 主程序 if __name__ == "__main__": # 示例股票代码 - 包含不同市场的股票 stock_codes = ["601288","000937","688360", "603300", "002195", "300014"] # 为每个股票代码获取并显示信息 for code in stock_codes: display_stock_info(code)输出的股票信息
PS C:\Users\czliu> & “C:/Program Files/Python313/python.exe” c:/Users/czliu/Documents/python/stocks_get.py
2025-07-17 18:26:35,917 – INFO – 正在请求股票数据: 601288
2025-07-17 18:26:36,043 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 601288
股票简称 XD农业银
总股本 349983033873.0
流通股 319244210777.0
行业 银行
总市值 2145395997641.48999
流通市值 1956967012063.01001
上市时间 20100715
最新价 6.13股票 601288 的数据已成功写入 get_stocks.xlsx
2025-07-17 18:26:36,282 – INFO – 正在请求股票数据: 000937
2025-07-17 18:26:36,385 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 000937
股票简称 冀中能源
总股本 3533546850.0
流通股 3433394152.0
行业 煤炭行业
总市值 20847926415.0
流通市值 20257025496.800003
上市时间 19990909
最新价 5.9股票 000937 的数据已成功写入 get_stocks.xlsx
2025-07-17 18:26:36,432 – INFO – 正在请求股票数据: 688360
2025-07-17 18:26:36,530 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 688360
股票简称 德马科技
总股本 263745667.0
流通股 252286035.0
行业 通用设备
总市值 5617782707.1
流通市值 5373692545.5
上市时间 20200602
最新价 21.3股票 688360 的数据已成功写入 get_stocks.xlsx
2025-07-17 18:26:36,582 – INFO – 正在请求股票数据: 603300
2025-07-17 18:26:36,678 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 603300
股票简称 海南华铁
总股本 1990650596.0
流通股 1990650596.0
行业 专用设备
总市值 23230892455.32
流通市值 23230892455.32
上市时间 20150529
最新价 11.67股票 603300 的数据已成功写入 get_stocks.xlsx
2025-07-17 18:26:36,716 – INFO – 正在请求股票数据: 002195
2025-07-17 18:26:36,813 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 002195
股票简称 岩山科技
总股本 5681544596.0
流通股 5612251612.0
行业 互联网服务
总市值 32043911521.439999
流通市值 31653099091.679996
上市时间 20071212
最新价 5.64股票 002195 的数据已成功写入 get_stocks.xlsx
2025-07-17 18:26:36,868 – INFO – 正在请求股票数据: 300014
2025-07-17 18:26:36,966 – INFO – 数据获取成功股票信息:
项目 数值
股票代码 300014
股票简称 亿纬锂能
总股本 2045721497.0
流通股 1861434646.0
行业 电池
总市值 92057467365.0
流通市值 83764559070.0
上市时间 20091030
最新价 45.0股票 300014 的数据已成功写入 get_stocks.xlsx
-
Python 的schedule库
#附赠通用封装模板(复制可用) import schedule import time from datetime import datetime def my_job(): print(f"[{datetime.now()}] 正在执行任务...") def run(): schedule.every().day.at("10:00").do(my_job) while True: schedule.run_pending() time.sleep(1) if __name__ == "__main__": run() -
股票量化对冲策略python代码
量化对冲策略计算绘图
代码提要
代码运行环境,WIN 11 ,python 3.13
- 使用
akshare获取股票数据,并用pandas的ewm函数 - 添加了
akshare库获取真实股票数据 - 主程序中使用
ak.stock_zh_index_daily获取上证指数数据 - 增加了异常处理,当获取数据失败时使用随机数据作为备选
- 确保数据列名与原代码兼容
- 截取最近200个交易日的数据进行计算
使用说明:
- 需要安装依赖库:
pip install pandas numpy akshare - 默认获取上证指数数据,可以通过修改
ak.stock_zh_index_daily(symbol="sh000001")中的symbol参数获取其他股票数据 - 股票代码格式示例:
- 上证指数: sh000001
- 深证成指: sz399001
- 创业板指: sz399006
- 个股: sh600000(浦发银行) 或 sz000001(平安银行)
- 本代码股票代码 sz000937 (冀中能源)
下面是添加了绘图功能的完整代码,使用
matplotlib绘制股票价格和各项量化指标的图表:import pandas as pd import numpy as np import akshare as ak import matplotlib.pyplot as plt import matplotlib.dates as mdates from matplotlib.ticker import MaxNLocator import matplotlib.patches as mpatches import warnings import matplotlib.font_manager as fm import os # 忽略警告 warnings.filterwarnings('ignore') # 设置全局字体 font_path = 'C:/Windows/Fonts/simhei.ttf' # 黑体字体路径,根据实际情况修改 if fm.findfont(fm.FontProperties(fname=font_path)): plt.rcParams['font.family'] = fm.FontProperties(fname=font_path).get_name() else: print("未找到指定字体,请检查路径。将使用默认字体。") # 解决负号显示问题 plt.rcParams['axes.unicode_minus'] = False def quant_hedge_strategy(df): # 主力暗盘资金副图指标 df['量化对冲1'] = df['close'].ewm(span=9, adjust=False).mean() df['量化对冲3'] = (df['量化对冲1'] * 1.14).ewm(span=5, adjust=False).mean() df['量化对冲4'] = df['close'].ewm(span=2, adjust=False).mean() # 计算130日高低点 df['130日高点'] = df['high'].rolling(130).max() df['130日低点'] = df['low'].rolling(130).min() # 计算条件柱状图 df['stickline_cond'] = np.where(df['量化对冲4'] >= df['量化对冲3'], 1, 0) # 计算其他指标 df['40日最低'] = df['low'].rolling(40).min() df['量化对冲2'] = (df['close'] - df['40日最低']) / df['40日最低'] * 100 df['MA40'] = df['close'].rolling(40).mean() df['量化对冲7'] = (df['close'] - df['MA40']) / df['MA40'] * 100 df['40日最低_MA40'] = df['40日最低'].rolling(40).mean() df['量化对冲8'] = (df['close'] - df['40日最低_MA40']) / df['40日最低_MA40'] * 100 df['量化对冲9'] = 28 df['量化对冲10'] = df['close'].rolling(5).mean() df['100日最高'] = df['量化对冲10'].rolling(100).max() df['量化对冲11'] = np.where(df['量化对冲10'] == df['100日最高'], df['量化对冲8'], np.nan) # 成本分布计算(简化为使用移动平均) df['cost90'] = df['close'].rolling(90).mean().shift(1) df['cost10'] = df['close'].rolling(10).mean().shift(1) df['量化对冲12'] = ((df['close'] * df['volume'] - df['cost90'] * df['volume']) / 20 + 250) * 1.2 / 5 df['量化对冲13'] = ((df['cost90'] * df['volume'] - df['close'] * df['volume']) / 20 + 250) * 1.2 / 5 df['量化对冲14'] = ((df['close'] * df['volume'] - df['cost10'] * df['volume']) / 20 + 250) * 1.2 / 5 df['量化对冲15'] = ((df['cost10'] * df['volume'] - df['close'] * df['volume']) / 20 + 250) * 1.2 / 5 # 主力暗盘资金副图指标信号条件 df['量化对冲5'] = ((df['量化对冲14'] > 0) & ((df['量化对冲12'] > df['量化对冲13']) | (df['量化对冲12'] > 0))) df['量化对冲6'] = ((df['量化对冲14'] > 0) & (df['量化对冲12'] > 0) & (df['量化对冲13'] < 0) & (df['量化对冲15'] < 0)) df['量化对冲5_count'] = df['量化对冲5'].rolling(10).sum() df['量化对冲6_count'] = df['量化对冲6'].rolling(60).sum() df['量化对冲16'] = ((df['量化对冲5'] & (df['量化对冲5_count'] == 1)) | (df['量化对冲6'] & (df['量化对冲6_count'] == 1))) df['量化对冲16_count'] = df['量化对冲16'].rolling(20).sum() df['起飞信号'] = np.where(df['量化对冲16'] & (df['量化对冲16_count'] == 1), df['量化对冲2'] * 1.16, np.nan) # 妖股信号 df['量化对冲14_60日高点'] = df['量化对冲14'].rolling(60).max() df['量化对冲14_20日高点'] = df['量化对冲14'].rolling(20).max() df['妖股条件1'] = df['量化对冲14'] >= df['量化对冲14_60日高点'] df['妖股条件2'] = (df['量化对冲14'] >= df['量化对冲14_20日高点']).rolling(20).sum() == 1 df['妖股条件3'] = df['量化对冲14'] > df['量化对冲13'] df['妖股信号'] = np.where(df['妖股条件1'] & df['妖股条件2'] & df['妖股条件3'], df['量化对冲2'] * 1.16, np.nan) # 交叉信号 df['交叉信号'] = np.where(df['量化对冲2'].shift(1) < df['量化对冲9'], df['量化对冲2'], np.nan) return df def plot_quant_strategy(df, stock_name="上证指数"): """绘制量化对冲策略图表""" # 创建一个包含4个子图的图表 fig, axes = plt.subplots(4, 1, figsize=(16, 18), sharex=True) fig.suptitle(f'{stock_name}量化对冲策略分析', fontsize=16) # 绘制K线图和均线 ax1 = axes[0] ax1.set_title('价格走势与均线', fontsize=14) ax1.plot(df.index, df['close'], label='收盘价', color='black', linewidth=2) ax1.plot(df.index, df['MA40'], label='MA40', color='blue', linestyle='--') ax1.plot(df.index, df['量化对冲10'], label='MA5', color='red', linestyle='--') # 绘制起飞信号 signal_dates = df[~df['起飞信号'].isna()].index signal_values = df[~df['起飞信号'].isna()]['close'] ax1.scatter(signal_dates, signal_values, color='green', s=100, label='起飞信号', marker='^') # 绘制妖股信号 妖股_dates = df[~df['妖股信号'].isna()].index 妖股_values = df[~df['妖股信号'].isna()]['close'] ax1.scatter(妖股_dates, 妖股_values, color='purple', s=100, label='妖股信号', marker='*') # 绘制交叉信号 cross_dates = df[~df['交叉信号'].isna()].index cross_values = df[~df['交叉信号'].isna()]['close'] ax1.scatter(cross_dates, cross_values, color='orange', s=100, label='交叉信号', marker='o') ax1.grid(True) ax1.legend(loc='upper left') ax1.set_ylabel('价格') # 绘制量化对冲2、9、11指标 ax2 = axes[1] ax2.set_title('量化对冲指标2、9、11', fontsize=14) ax2.plot(df.index, df['量化对冲2'], label='量化对冲2', color='blue') ax2.plot(df.index, df['量化对冲9'], label='量化对冲9', color='red', linestyle='--') ax2.plot(df.index, df['量化对冲11'], label='量化对冲11', color='green', marker='o', linestyle='', alpha=0.7) # 填充量化对冲2大于量化对冲9的区域 ax2.fill_between(df.index, df['量化对冲2'], df['量化对冲9'], where=(df['量化对冲2'] > df['量化对冲9']), color='lightgreen', alpha=0.3) # 填充量化对冲2小于量化对冲9的区域 ax2.fill_between(df.index, df['量化对冲2'], df['量化对冲9'], where=(df['量化对冲2'] < df['量化对冲9']), color='lightcoral', alpha=0.3) ax2.grid(True) ax2.legend(loc='upper left') ax2.set_ylabel('指标值') # 绘制量化对冲12、13、14、15指标 ax3 = axes[2] ax3.set_title('主力暗盘资金指标', fontsize=14) ax3.plot(df.index, df['量化对冲12'], label='量化对冲12', color='blue') ax3.plot(df.index, df['量化对冲13'], label='量化对冲13', color='red') ax3.plot(df.index, df['量化对冲14'], label='量化对冲14', color='green') ax3.plot(df.index, df['量化对冲15'], label='量化对冲15', color='purple') # 绘制起飞信号和妖股信号在这张图上的位置 ax3.scatter(signal_dates, df.loc[signal_dates, '量化对冲14'], color='green', s=50, marker='^') ax3.scatter(妖股_dates, df.loc[妖股_dates, '量化对冲14'], color='purple', s=50, marker='*') ax3.grid(True) ax3.legend(loc='upper left') ax3.set_ylabel('资金指标') # 绘制量化对冲5、6指标柱状图 ax4 = axes[3] ax4.set_title('量化对冲信号5、6', fontsize=14) # 绘制柱状图 bar_width = 0.6 quant5_positive = df['量化对冲5_count'].copy() quant5_positive[quant5_positive < 0] = 0 quant6_positive = df['量化对冲6_count'].copy() quant6_positive[quant6_positive < 0] = 0 ax4.bar(df.index, quant5_positive, width=bar_width, color='blue', alpha=0.7, label='量化对冲5') ax4.bar(df.index, quant6_positive, width=bar_width, color='green', alpha=0.7, label='量化对冲6') # 添加信号标记 ax4.scatter(signal_dates, [0]*len(signal_dates), color='green', s=100, marker='^') ax4.scatter(妖股_dates, [0]*len(妖股_dates), color='purple', s=100, marker='*') ax4.grid(True) ax4.legend(loc='upper left') ax4.set_ylabel('信号强度') ax4.set_xlabel('日期') # 设置x轴日期格式 locator = mdates.AutoDateLocator() formatter = mdates.ConciseDateFormatter(locator) for ax in axes: ax.xaxis.set_major_locator(locator) ax.xaxis.set_major_formatter(formatter) plt.tight_layout() plt.subplots_adjust(top=0.94) return fig # 使用示例 if __name__ == "__main__": # 使用akshare获取股票数据 try: # 获取上证指数数据,可替换为其他股票代码 stock_data = ak.stock_zh_index_daily(symbol="sz000937") # 重命名列以匹配原代码 df = stock_data.rename(columns={ 'date': 'date', 'open': 'open', 'high': 'high', 'low': 'low', 'close': 'close', 'volume': 'volume' }) # 设置日期索引 df['date'] = pd.to_datetime(df['date']) df.set_index('date', inplace=True) # 确保数据量足够 if len(df) < 200: print(f"数据长度不足,只有{len(df)}行") else: # 截取最近200个交易日数据 df = df.iloc[-200:] # 计算量化对冲指标 result_df = quant_hedge_strategy(df) # 绘制图表 fig = plot_quant_strategy(result_df, "冀中能源") # 显示图表 plt.show() # 输出结果 print(result_df[['量化对冲1', '量化对冲2', '量化对冲3', '量化对冲4', '量化对冲11', '起飞信号', '妖股信号', '交叉信号']].tail()) except Exception as e: print(f"获取数据出错: {e}") print("使用随机数据作为替代") # 随机数据作为备选 data = { 'date': pd.date_range(start='2023-01-01', periods=200), 'open': np.random.uniform(100, 200, 200), 'high': np.random.uniform(110, 220, 200), 'low': np.random.uniform(90, 190, 200), 'close': np.random.uniform(100, 200, 200), 'volume': np.random.randint(10000, 100000, 200) } df = pd.DataFrame(data) df.set_index('date', inplace=True) # 计算量化对冲指标 result_df = quant_hedge_strategy(df) # 绘制图表 fig = plot_quant_strategy(result_df, "随机数据") # 显示图表 plt.show() # 输出结果 print(result_df[['量化对冲1', '量化对冲2', '量化对冲3', '量化对冲4', '量化对冲11', '起飞信号', '妖股信号', '交叉信号']].tail())新增的绘图功能特点:
- 包含4个子图,全面展示量化策略的各项指标:
- 子图1:股票价格走势与关键均线,标记各种信号出现位置
- 子图2:量化对冲2、9、11指标,填充指标间区域
- 子图3:主力暗盘资金指标(量化对冲12-15)
- 子图4:量化对冲信号5、6的柱状图
- 信号标记:
- 绿色三角:起飞信号
- 紫色星号:妖股信号
- 橙色圆点:交叉信号
- 图表美化:
- 自动调整日期显示格式
- 添加网格线提高可读性
- 每个子图都有标题和图例
- 合理的颜色搭配
- 交互功能:
- 鼠标悬停可查看具体数值
- 可放大缩小特定区域
- 可保存图表为图片文件
使用说明:
- 需要安装额外依赖库:
pip install matplotlib - 运行代码后会自动显示绘制的图表
- 图表标题会显示股票名称,默认为”上证指数”
- 可以通过修改
plot_quant_strategy函数来自定义图表样式和内容
量化对冲指标的解释
以下是代码中主要量化对冲指标的解释,这些指标用于分析股票趋势、资金流向和潜在交易信号:
1. 趋势跟踪指标
量化对冲1
- 计算方式:
EMA(close, 9)(9日指数移动平均线) - 作用:反映短期价格趋势,比简单均线更敏感,用于识别短期方向。
量化对冲3
- 计算方式:
EMA(量化对冲1 × 1.14, 5)(对放大后的短期均线再做5日EMA) - 作用:增强趋势信号,放大短期波动,用于捕捉快速变化的趋势。
量化对冲4
- 计算方式:
EMA(close, 2)(2日指数移动平均线) - 作用:超短期趋势指标,对价格变动极为敏感,用于识别极短期转折点。
2. 相对强弱指标
量化对冲2
- 计算方式:
(close - 40日最低) / 40日最低 × 100 - 作用:衡量当前价格相对于近40日最低价的涨幅百分比,类似于RSI指标,反映价格强度。
- 信号逻辑:值越高,表明价格距离低点越远,可能处于超买区域;反之可能处于超卖区域。
量化对冲7
- 计算方式:
(close - MA40) / MA40 × 100 - 作用:衡量当前价格相对于40日均价的偏离程度,用于判断价格是否偏离均值过远(可能回调)。
3. 资金流向指标
量化对冲12-15
- 计算方式:基于收盘价、成交量与不同周期成本价(90日/10日均价)的差值计算
量化对冲12 = ((close × volume - cost90 × volume) / 20 + 250) × 1.2 / 5量化对冲13 = ((cost90 × volume - close × volume) / 20 + 250) × 1.2 / 5量化对冲14 = ((close × volume - cost10 × volume) / 20 + 250) × 1.2 / 5量化对冲15 = ((cost10 × volume - close × volume) / 20 + 250) × 1.2 / 5- 作用:模拟不同周期的资金流入流出情况,正值表示资金流入,负值表示资金流出。
- 信号逻辑:当短期资金流入(如量化对冲14)强于长期资金流出(如量化对冲13)时,视为看涨信号。
4. 信号生成指标
起飞信号
- 触发条件:
量化对冲16为True(满足量化对冲5或量化对冲6条件)- 且
量化对冲16在近20天内首次出现
- 作用:捕捉价格启动初期的信号,可能预示短期快速上涨。
妖股信号
- 触发条件:
量化对冲14 ≥ 量化对冲14_60日高点(短期资金流入创60日新高)- 且
量化对冲14 ≥ 量化对冲14_20日高点在近20天内仅出现1次 - 且
量化对冲14 > 量化对冲13(短期资金流入强于长期流出)
- 作用:识别可能出现大幅上涨的”妖股”特征,通常伴随强资金推动。
交叉信号
- 触发条件:
量化对冲2的前一日值低于量化对冲9(阈值28),当日值高于或等于阈值 - 作用:捕捉价格从弱势区域向上突破的转折点,类似MACD金叉信号。
5. 辅助指标
量化对冲5 & 量化对冲6
- 量化对冲5:短期资金流入为正,且(短期资金流入>长期资金流出 或 短期资金流入为正)
- 量化对冲6:短期资金流入为正,且短期和长期资金流入均为正,同时短期和长期资金流出均为负
- 作用:量化对冲5侧重短期资金流向,量化对冲6要求更严格的资金面条件,两者结合用于确认强势信号。
策略逻辑总结
- 趋势判断:通过多周期均线(量化对冲1、3、4)识别价格趋势方向。
- 强弱评估:通过量化对冲2评估价格相对强度,结合阈值(量化对冲9)判断超买超卖。
- 资金监控:通过量化对冲12-15跟踪不同周期资金流向,判断多空力量对比。
- 信号生成:综合上述指标,生成起飞信号、妖股信号和交叉信号,作为交易决策依据。
该策略适合用于识别短期趋势转折点和强势股,但实际应用中需结合止损机制和仓位管理,避免假信号带来的风险。

- 使用
-
python计算股票年化收益率、年化波动率、夏普比率系数(>1,good)并绘图
python代码,由vscode trae插件生成–无需任何调整,即可运行
import akshare as ak import pandas as pd import numpy as np import matplotlib.pyplot as plt def get_stock_data(stock_code, start_date, end_date): """ 获取指定股票的历史数据 :param stock_code: 股票代码,例如 'sh600519' :param start_date: 开始日期,格式 'YYYYMMDD' :param end_date: 结束日期,格式 'YYYYMMDD' :return: 包含历史数据的 DataFrame """ stock_df = ak.stock_zh_a_hist(symbol=stock_code[2:], period="daily", start_date=start_date, end_date=end_date) stock_df['日期'] = pd.to_datetime(stock_df['日期']) stock_df.set_index('日期', inplace=True) return stock_df def calculate_risk_metrics(stock_df): """ 计算股票风险指标 :param stock_df: 包含历史数据的 DataFrame :return: 包含风险指标的字典和包含日收益率的 DataFrame """ # 计算日收益率 stock_df['日收益率'] = stock_df['收盘'].pct_change() # 计算年化收益率 annual_return = stock_df['日收益率'].mean() * 252 # 计算年化波动率(收益率标准差) annual_volatility = stock_df['日收益率'].std() * np.sqrt(252) # 假设无风险利率为 2% risk_free_rate = 0.02 # 计算夏普比率 sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility return { '年化收益率': annual_return, '年化波动率': annual_volatility, '夏普比率': sharpe_ratio }, stock_df def plot_stock_data(stock_df, stock_code): """ 绘制股价变化和日收益率变化图 :param stock_df: 包含历史数据和日收益率的 DataFrame :param stock_code: 股票代码 """ # 设置图片清晰度 plt.rcParams['figure.dpi'] = 300 # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False # 创建一个包含两个子图的画布 fig, axes = plt.subplots(2, 1, figsize=(12, 8)) # 绘制股价变化图 axes[0].plot(stock_df['收盘'], label='收盘价') axes[0].set_title(f'{stock_code} 股价变化') axes[0].set_ylabel('价格') axes[0].legend() # 绘制日收益率变化图 axes[1].plot(stock_df['日收益率'], label='日收益率', color='orange') axes[1].set_title(f'{stock_code} 日收益率变化') axes[1].set_xlabel('日期') axes[1].set_ylabel('收益率') axes[1].legend() # 手动调整子图之间的垂直间距 plt.subplots_adjust(hspace=0.5) plt.show() if __name__ == "__main__": stock_code = 'sh600938' # 农业银行 start_date = '20240101' end_date = '20250531' # 获取股票数据 stock_data = get_stock_data(stock_code, start_date, end_date) # 计算风险指标 risk_metrics, stock_data = calculate_risk_metrics(stock_data) print(f"股票代码: {stock_code}") print(f"开始日期: {start_date}") print(f"结束日期: {end_date}") print("风险指标:") for key, value in risk_metrics.items(): print(f"{key}: {value:.4f}") # 绘制图表 plot_stock_data(stock_data, stock_code)计算结果:农行、中国海油
股票代码: sh600938 中国海油
开始日期: 20240101
结束日期: 20250531
风险指标:
年化收益率: 0.2109
年化波动率: 0.3317 波动大、影响收益
夏普比率: 0.5756股票代码: sh601288 农业银行
开始日期: 20240101
结束日期: 20250531
风险指标:
年化收益率: 0.3293
年化波动率: 0.2067 波动小收益高
夏普比率: 1.4961 收益高 -
使用Python和akshare库分析股票风险
要分析某只股票的风险,我们可以使用akshare库获取股票数据,然后进行风险指标计算绘图。以下是一个完整的Python代码示例:
import akshare as ak import pandas as pd import numpy as np import matplotlib.pyplot as plt from scipy.stats import norm from datetime import datetime, timedelta # 设置中文显示 plt.rcParams['font.sans-serif'] = ['SimHei'] plt.rcParams['axes.unicode_minus'] = False def get_stock_data(stock_code, start_date, end_date): """ 获取股票历史数据 """ try: # 使用akshare获取股票数据 df = ak.stock_zh_a_hist(symbol=stock_code, period="daily", start_date=start_date, end_date=end_date, adjust="hfq") df['日期'] = pd.to_datetime(df['日期']) df.set_index('日期', inplace=True) df.sort_index(inplace=True) return df except Exception as e: print(f"获取数据失败: {e}") return None def calculate_risk_metrics(df): """ 计算风险指标 """ if df is None or len(df) < 30: print("数据不足或获取失败") return None # 计算日收益率 df['日收益率'] = df['收盘'].pct_change() # 计算风险指标 metrics = {} # 1. 波动率(年化) daily_volatility = df['日收益率'].std() annual_volatility = daily_volatility * np.sqrt(252) metrics['年化波动率'] = annual_volatility # 2. 最大回撤 df['累计收益率'] = (1 + df['日收益率']).cumprod() df['前高'] = df['累计收益率'].cummax() df['回撤'] = (df['累计收益率'] - df['前高']) / df['前高'] max_drawdown = df['回撤'].min() metrics['最大回撤'] = max_drawdown # 3. VaR (95%置信度) var_95 = norm.ppf(0.05, df['日收益率'].mean(), df['日收益率'].std()) metrics['日VaR(95%)'] = var_95 # 4. 夏普比率(假设无风险利率为3%) annual_return = (1 + df['日收益率'].mean()) ** 252 - 1 risk_free_rate = 0.03 sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility metrics['夏普比率'] = sharpe_ratio # 5. Beta系数(需要市场数据,这里简化处理) # 实际应用中需要获取市场指数数据计算 return metrics def plot_stock_risk(df): """ 绘制风险相关图表 """ if df is None or len(df) < 30: return plt.figure(figsize=(15, 10)) # 1. 价格走势 plt.subplot(2, 2, 1) plt.plot(df['收盘'], label='收盘价') plt.title('股票价格走势') plt.xlabel('日期') plt.ylabel('价格') plt.legend() # 2. 收益率分布 plt.subplot(2, 2, 2) plt.hist(df['日收益率'].dropna(), bins=50, density=True, alpha=0.6, color='g') plt.title('日收益率分布') plt.xlabel('日收益率') plt.ylabel('频率') # 3. 累计收益率和回撤 plt.subplot(2, 2, 3) plt.plot(df['累计收益率'], label='累计收益率') plt.plot(df['前高'], label='前高', linestyle='--') plt.title('累计收益率与回撤') plt.xlabel('日期') plt.ylabel('累计收益率') plt.legend() # 4. 回撤曲线 plt.subplot(2, 2, 4) plt.plot(df['回撤'], label='回撤', color='r') plt.title('回撤曲线') plt.xlabel('日期') plt.ylabel('回撤比例') plt.legend() plt.tight_layout() plt.show() def analyze_stock_risk(stock_code, start_date, end_date): """ 分析股票风险的主函数 """ print(f"正在分析股票 {stock_code} 的风险...") # 获取数据 df = get_stock_data(stock_code, start_date, end_date) if df is None: return # 计算风险指标 metrics = calculate_risk_metrics(df) if metrics: print("\n风险指标分析结果:") for key, value in metrics.items(): print(f"{key}: {value:.4f}") # 绘制图表 plot_stock_risk(df) else: print("无法计算风险指标") # 示例使用 if __name__ == "__main__": # 设置股票代码和时间范围 stock_code = "601288" # 平安银行,可以替换为其他股票代码 # 计算两年前的日期 start_date = (datetime.now() - timedelta(days=365*1)).strftime("%Y%m%d") # 获取当前日期 end_date = datetime.now().strftime("%Y%m%d") analyze_stock_risk(stock_code, start_date, end_date)代码说明
- 数据获取:
- 使用akshare的
“stock_zh_a_hist”函数获取A股历史数据 - 数据包括日期、开盘价、收盘价、最高价、最低价等
- 风险指标计算:
- 年化波动率: 反映股票价格的波动程度
- 最大回撤: 衡量从最高点到最低点的最大损失
- VaR(风险价值): 在95%置信度下的日最大损失
- 夏普比率: 衡量风险调整后的收益
- 可视化:
- 价格走势图
- 收益率分布直方图
- 累计收益率和前高曲线
- 回撤曲线
使用说明
- 安装所需库:
“pip install akshare pandas numpy matplotlib scipy” - 修改
“stock_code”为你想分析的股票代码(如”600519″为贵州茅台) - 调整
“start_date”和
“end_date”设置分析的时间范围 - 运行代码即可获取风险分析结果和可视化图表
注意事项
- akshare的数据接口可能会有变化,如果获取失败请检查akshare文档
- Beta系数的计算需要市场指数数据,这里简化处理未实现
- 更全面的风险分析可以添加更多指标,如下行风险、索提诺比率等
希望这个代码能帮助你分析股票风险!如需进一步扩展,可以考虑添加更多风险指标或与其他金融数据的关联分析。

-
NiceGUI 是一个基于 Python 的现代 Web 应用框架
NiceGUI 是一个基于 Python 的现代 Web 应用框架,它允许开发者直接使用 Python 构建交互式 Web 界面,而无需编写前端代码。以下是 NiceGUI 的主要功能和特点:
核心功能
1.简单易用的 UI 组件 提供按钮、文本框、下拉菜单、滑块、图表等常见 UI 元素 支持布局管理(水平、垂直、卡片式等) 可自定义样式和交互行为
2. 实时交互 支持双向数据绑定 事件驱动编程模型(点击、输入变化等) 实时更新 UI 内容
3. 数据可视化 集成 Plotly、Matplotlib 等图表库 支持实时数据图表 可展示 3D 图形和复杂数据可视化
4. 多平台支持 可以作为 Web 应用运行 也可以打包为桌面应用(通过 PyInstaller 等工具) 支持移动设备访问
5. 扩展性 可集成自定义 JavaScript 代码 支持与其他 Python 库(如 Pandas、NumPy)无缝协作 可扩展自定义组件示例代码
下面是一个简单的 NiceGUI 应用示例,展示了基本的 UI 组件和交互功能: UI组件可输出Markdown、HTML、可运行script脚本
from nicegui import ui, Client name = ui.input(label='你的名字') button = ui.button('打招呼', on_click=lambda: result.set_text(f'你好,{name.value}!')) result = ui.label() # 嵌入JavaScript def run_js(): # Execute JavaScript code ui.run_javascript('alert("这是来自JavaScript的弹窗!")') # 嵌入JavaScript def get_window_width(): # Execute JavaScript to get window width width = ui.run_javascript('return window.innerWidth') ui.notify(f'窗口宽度: {width}px') with ui.row(): ui.button('运行JS', on_click=run_js) ui.button('获取窗口宽度', on_click=get_window_width) # 嵌入HTML内容 html_content = """ <div style="background-color:#f0f0f0; padding:10px; border-radius:5px;"> <h3>HTML内容展示</h3> <p>这是一段嵌入的<strong>HTML</strong>代码</p> <button onclick="alert('HTML按钮被点击!')",color="blue">HTML按钮</button> </div> """ ui.html(html_content) # 嵌入Markdown内容 markdown_content = """ # Markdown示例 这是一个展示**Markdown**功能的示例: 1. 列表项1 2. 列表项2 3. 列表项3 > 这是一段引用 [NiceGUI官网](https://nicegui.io) 代码示例: ```python def hello_world(): print("Hello from Markdown!") aaa bbb ccc 表格1: |姓名 | 年龄 |城市| |---|---|---| |张三| 25 | 北京| |李四 |30 | 上海| 表格2: | 姓名 | 年龄 | 职业 | | --- | --- | --- | | 张三 | 28 | 工程师 | | 李四 | 32 | 设计师 | | 王五 | 45 | 产品经理 | """ ui.markdown(markdown_content) ui.html('-------控件绑定属性-----------') 绑定属性 class Demo: def init(self): self.number = 1 demo = Demo() v = ui.checkbox('visible', value=True) with ui.column().bind_visibility_from(v, 'value'): ui.slider(min=1, max=3).bind_value(demo, 'number') ui.toggle({1: 'A', 2: 'B', 3: 'C'}).bind_value(demo, 'number') ui.number().bind_value(demo, 'number') ui.run()应用场景
数据仪表板:实时显示和分析数据
交互式工具:构建科学计算、工程或数据分析工具
快速原型开发:无需前端知识即可快速构建 Web 应用原型
桌面应用替代方案:使用 Web 技术构建跨平台桌面应用
NiceGUI 特别适合 Python 开发者快速构建功能丰富的 Web 界面,尤其在数据科学和自动化领域有很大
