Python办公自动化之 - Excel

前言

一、概述

二、方法速查

# 导包
from openpyxl import load_workbook
from openpyxl.styles import Font, PatternFill, Border, Side, Alignment

2.1、文档对象和sheet

### 文档对象和sheet
workbook = load_workbook(filename = "./test.xlsx")   # 实例化workbook对象；如果不输入路径，在save时输入路径则会新建一个Excel文件
workbook.create_sheet("new_sheet")   # 创建新的 sheet
workbook.remove("new_sheet")         # 删除一个 sheet

sheet_list = workbook.sheetnames   # 获取 Excel 中所有的 sheet 名称，返回列表
sheet = workbook[sheet_list[0]]    # 实例化 Excel 的第一个 sheet 对象
sheet.title = 'new_name'           # 修改 sheet 名称
sheet = workbook.active            # 实例化活跃的 sheet 对象（就是你用Office打开Excel时被打开的sheet）

2.2、单元格定位和数据

### 读取
## 定位单个cell，获取数据
cell_A1 = sheet["A1"]                  # 坐标定位单元格
cell_A1 = sheet.cell(row=1,column=1)   # 行列定位单元格

cell_value = cell_A1.value             # 获取单元格数据
cell_row = cell_A1.row                 # 获取单元格所在行
cell_column = cell_A1.column           # 获取单元格所在列
cell_coordinate = cell_A1.coordinate   # 获取单元格坐标，Excel表达如：A1

cell_A1 = '我是写入的内容'                # 向单元格写入数据


## 定位范围
cells = sheet["A1:C2"]   # 获取A1到C2范围的单元格
cells = sheet["A"]       # 获取A列的数据
cells = sheet["A:C"]     # 获取A,B,C三列的数据
cells = sheet[5]         # 获取第5行的数据

cells = sheet.iter_rows(min_row=2, max_row=5, min_col=1, max_col=2)   # 定位第二行到第五行，第一列到第二列的范围，一行一行的读
cells = sheet.iter_cols(min_row=2, max_row=5, min_col=1, max_col=2)   # 定位第二行到第五行，第一列到第二列的范围，一列一列的读

rows = sheet.rows        # 获取所有的行
columns = sheet.columns  # 获取所有的列

2.3、追加、插入和删除

sheet.append(data)       # 将data追加到sheet已有数据的后面（按行追加），data的格式有list, range or generator, or dict
		append([‘This is A1’, ‘This is B1’, ‘This is C1’])   # 列表，需要所有列都有数据，如果某一列没有，可以占坑
		append({‘A’ : ‘This is A1’, ‘C’ : ‘This is C1’})     # 字典，通过列名来定位数据追加的位置
    append({1 : ‘This is A1’, 3 : ‘This is C1’})         # 字典，通过列索引来定位数据追加的位置，所以是从1开始的

sheet.insert_cols(idx=4,amount=2)   # 插入列，idx为插入的位置，amount为插入的数量，插入之后的列在原来的第4行的左边，可以理解为：插入一列作为第四列，只不过这里是插入多列
sheet.insert_rows(idx=5,amount=4)   # 插入行，idx为插入的位置，amount为插入的数量，插入之后的行在原来的第5行的上边

sheet.delete_cols(idx=4,amount=2)   # 删除列，从第四列开始，删除两列
sheet.delete_rows(idx=5,amount=4)   # 删除行，从第五行开始，删除四行

2.4、样式设置

# 字体样式
font = Font(name = 字体名称, size = 字体大小, bold = 是否加粗, italic = 是否斜体, color = 字体颜色)
font = Font(name="微软雅黑",size=20,bold=True,italic=True,color="FF0000")

# 对齐样式
alignment = Alignment(horizontal = 水平对齐模式, vertical = 垂直对齐模式, text_rotation = 旋转角度, wrap_text = 是否自动换行)
alignment = Alignment(horizontal="center",vertical="center",text_rotation=45,wrap_text=True)
	- 水平对齐：‘distributed’，‘justify’，‘center’，‘leftfill’， ‘centerContinuous’，‘right，‘general’；
	- 垂直对齐：‘bottom’，‘distributed’，‘justify’，‘center’，‘top’；

# 边框样式
border = Border(left = Side(style=边线样式，color=边线颜色)，right = 右边线样式，top = 上边线样式，bottom = 下边线样式)
style参数的种类： 'double, ‘mediumDashDotDot’, ‘slantDashDot’,‘dashDotDot’,‘dotted’,‘hair’,'mediumDashed, ‘dashed’, ‘dashDot’, ‘thin’,‘mediumDashDot’,‘medium’, 'thick’from openpyxl.styles import Side,Borderfrom
border = Border(
         left=Side(border_style='thin', color='000000'),
         right=Side(border_style='thin', color='000000'),
         top=Side(border_style='thin', color='000000'),
         bottom=Side(border_style='thin', color='000000'),
     )

# 填充样式
pattern_fill = PatternFill(fill_type = 填充样式，fgColor = 填充颜色）
pattern_fill = PatternFill(fill_type="solid",fgColor="99ccff")
GradientFill(stop=(渐变颜色 1，渐变颜色 2……))
gradient_fill = GradientFill(stop=("FFFFFF","99ccff","000000"))


cell.font = font             # 设置字体
cell.border = border         # 设置边框
cell.fill = pattern_fill     # 设置常规填充
cell.fill = gradient_fill    # 设置渐变填充
cell.alignment = alignment   # 设置居中对齐
            

# 行高样式
sheet.column_dimensions[1].height = 8     # 一行设置
sheet.column_dimensions['A'].width = 8    # 一列设置
sheet.row_dimensions.height = 50          # 整体设置
sheet.column_dimensions.width = 30        # 整体设置

2.5、其他设置

# 其他常用设置
sheet.dimensions                           # sheet中有内容的范围，返回Excel中的范围表达，如：'A1:H11'
sheet.freeze_panes = "C3"                  # 冻结窗口
sheet.auto_filter.ref = "A1"               # 给表格添加 “筛选器”
sheet.auto_filter.ref = sheet.dimensions   # 给所有列加筛选器

# 合并单元格
sheet.merge_cells("C1:D2")
sheet.merge_cells(start_row=7,start_column=1,end_row=8,end_column=3)

2.6、使用Excel公式

for i in range(2,16):
    sheet[f"D{i}"] = f'=IF(RIGHT(C{i},2)="cm",C{i},SUBSTITUTE(C{i},"m","")*100&"cm")'

三、实践和问题

import os
import re
import json
import time
import warnings
import requests
import datetime
# import numpy as np
# import pandas as pd
from openpyxl import load_workbook


warnings.filterwarnings('ignore')


# 浏览器头
headers = {'content-type': 'application/json; charset=utf-8',
           'User-Agent': 'xxx'}   # 请求头自行填写


# 获取预估数据 或 历史数据
def get_data(url, is_Fstimate):
    """_summary_
    Args:
        url (_type_): 每个基金的URL,可能是今天的估值数据,也可能是当前基金所有的历史数据,通过is_Fstimate参数来区分
        is_Fstimate (bool): 用来区分基金 url 类型,1 为估值数据,0 为历史数据

    Returns:
        _type_: 返回处理好的数据格式
    """
    r = requests.get(url, headers=headers)
    # 返回信息
    content = r.text
    # print(url, content)
    
    if is_Fstimate:            
        # 正则表达式
        pattern = r'^jsonpgz\((.*)\)'
        # 查找结果
        search = re.findall(pattern, content)
        data = json.loads(search[0])
    else:
        # 正则表达式
        pattern = r'Data_netWorthTrend = \[(.*?)\]'
        # 查找结果
        search = re.findall(pattern, content)
        # search的形状是['{},{},{},{}']，但是loads能处理的形状是：1、json中只有一个字典的情况：'{}' 或者 2、json中只有多个字典的情况'[{},{},{},{}]'
        # print(search)
        # print(search[0])
        search = '[' + search[0] + ']'
        data = json.loads(search)
        data = data[::-1]   # 列表反转
    
    return  data


# 时间转换函数
def timeTransformation(datetime):
    """_summary_

    Args:
        datetime (_type_): 数据日期,当前为时间戳,粗腰处理成日期

    Returns:
        _type_: 处理好的日期
    """
    datelocal = time.localtime(datetime)
    datetime_ = time.strftime("%Y-%m-%d %H:%M:%S", datelocal)
    date = datetime_[:10]
    
    return date


# 写入历史数据
def write_history_data(history_data, sheet, all_date=None, omit_date_len=0, is_rewirte=0):
    """_summary_

    Args:
        history_data (_type_): api 中获取到的历史数据
        sheet (_type_): 当前基金对应的 sheet
        omit_date_len (_type_):  是否有断层数据
        is_rewirte (int, optional): 是否需要重写所有数据,可能导致数据错位,慎重选择. Defaults to 0.
    """
    if is_rewirte==1:   # 完全重写历史数据,买入信息可能会错位,慎重选择
        for r in range(len(history_data)):
            date_r = timeTransformation(int(history_data[r]['x']/1000))   
            sheet.cell(row=r+3, column=1).value = date_r   # 日期
            sheet.cell(row=r+3, column=2).value = history_data[r]['y']   # 净值
            sheet.cell(row=r+3, column=3).value = history_data[r]['equityReturn']   # 涨幅
    else:   
        if omit_date_len>0:   # 如果断层数据大于0天,则将断层数据插入表中(一般这个值小于等于 0)
            for m in range(omit_date_len):
                sheet.insert_rows(idx=3+m, amount=1)
                date_m = timeTransformation(int(history_data[m]['x']/1000))  
                sheet.cell(row=3+m, column=1).value = date_m   # 日期
                sheet.cell(row=3+m, column=2).value = history_data[m]['y']   # 净值
                sheet.cell(row=3+m, column=3).value = history_data[m]['equityReturn']   # 涨幅
        else:
            # 将昨天的预估数据更新为实际历史数据
            latestDate = timeTransformation(int(history_data[0]['x']/1000))
            if latestDate not in all_date:
                sheet.insert_rows(idx=3, amount=1)
                row = 3
            else:
                is_A = [i.value for i in sheet['A']]   # 获取A列数据
                is_A_index = [i for i,v in enumerate(is_A) if v==latestDate]   # 获取A列时间为昨天的行的索引
                row = is_A_index[0] + 1
            sheet.cell(row=row, column=1).value = timeTransformation(int(history_data[0]['x']/1000))   # 日期
            sheet.cell(row=row, column=2).value = history_data[0]['y']   # 净值
            sheet.cell(row=row, column=3).value = history_data[0]['equityReturn']   # 涨幅
            
            
# 写入预估数据
def write_estimat_data(estimate_data, sheet, all_date):
    # 如果不用重新写入，则需要将最新的历史数据和预估数据插入到表中
    if estimate_data['gztime'][:10] not in all_date:
        sheet.insert_rows(idx=3, amount=1)
    # 预估数据插入
    sheet.cell(row=3, column=1).value = estimate_data['gztime'][:10]
    sheet.cell(row=3, column=2).value = float(estimate_data['gsz'])
    sheet.cell(row=3, column=3).value = float(estimate_data['gszzl'])
    # print(estimate_data['gztime'][:10],float(estimate_data['gsz']),float(estimate_data['gszzl']))


# 计算累计涨跌幅(从最高点的跌幅)
def whetherNotToBuy(sheet):
    ## 获取上次购买数据所在的行,判断购买的规则为:建议购买==1 and 购买金额不为0
    is_F = [i.value for i in sheet['F']]   # 获取E列数据
    is_G = [i.value for i in sheet['G']]
    is_F_index = [i for i,v in enumerate(is_F) if v==1 and is_G[i] != None]   # 获取买入点的数据索引
    if len(is_F_index)<1:   
        return
    else:
        row = is_F_index[0]   # 上次购买所在的行数的前一行
    
    ## 计算买入之后累计涨跌幅度
    for i in range(row-2): # 减去两行表头
        sheet[f'D{row-i}'] = sum([j.value for j in sheet['C'][row-i-1:row]])  # row-i-1是由于sheet中切片是前闭后开的，所以要减去一行
    
    ## 计算买入之后从最高点的跌幅
    B_list = [i.value for i in sheet['B'][2:row]]   # 获取自上次买入后的净值数据,sheet中切片是前闭后开的,这里相当于是从第3行取到第row行
    # 获取最大净值所在的行号
    if len(B_list)==0:   
        max_row = 3
    else:
        max_row = B_list.index(max(B_list))+3
    # 最大值行号为 3 的话,说明上一天才买入
    if max_row==3:
        pass
    else:
        for i in range(1,max_row-2): # 减去两行表头
            sheet[f'E{max_row-i}'] = sum([j.value for j in sheet['C'][max_row-i-1:max_row-1]])  # row-i-1是由于sheet中切片是前闭后开的，所以要减去一行
            if sheet[f'E{max_row-i}'].value >= -4:
                sheet[f'F{max_row-i}'] = 0
            else:
                sheet[f'F{max_row-i}'] = 1
        

def invoking_func(codes,workbook):
    sheetnames = workbook.sheetnames
    for code in codes:
        # 生成链接
        url_estimate = "http://fundgz.1234567.com.cn/js/%s.js"%code
        url_history = "http://fund.eastmoney.com/pingzhongdata/%s.js"%code
        
        # 获取数据
        estimate_data = get_data(url_estimate, True)   # 前一天的更新数据和当天的预估数据
        history_data = get_data(url_history, False)    # 所有历史数据
        
        ##### 数据入表 #####
        # 如果 sheet 不存在,创建sheet并写入数据
        if estimate_data['name'] not in sheetnames:
            workbook.create_sheet(estimate_data['name'])
            sheet = workbook[estimate_data['name']]
            
            # 表头写入
            cells = ['A1','A2','B2','C2','D2','E2','F2','G2']
            columns = ['基金代码：' + code, '交易日期','单位净值','净值涨跌','自上次买入累计涨跌','自买入后最大值累计跌幅','是否建议买入','买入金额']
            for p in zip(cells,columns):
                sheet[p[0]] = p[1]  
            # 写入所有历史数据
            write_history_data(history_data, sheet,is_rewirte=1)
            # 写入预估数据
            all_date = [sheet['A'][a].value for a in range(len(sheet['A']))][2:]   # 更新表中的时间数据,否则会把最新的预估数据和最新的历史数据都写进去
            write_estimat_data(estimate_data, sheet, all_date)
            
        else:
            sheet = workbook[estimate_data['name']]
            # 表中已有的数据的所有日期
            all_date = [sheet['A'][a].value for a in range(len(sheet['A']))][2:]
            
            # 写入历史数据 (完全重写和更新最新历史数据(昨天数据),完全重写会导致买入信息错位,慎重选择)
            is_rewirte = 0   # 是否需要完全重写数据
            omit_date_len = len(history_data) - len(all_date)   # 数据缺失天数
            write_history_data(history_data, sheet, all_date, omit_date_len, is_rewirte=0)
            
            # 写入预估数据
            all_date = [sheet['A'][a].value for a in range(len(sheet['A']))][2:]   # 更新表中的时间数据,否则会把最新的预估数据和最新的历史数据都写进去
            write_estimat_data(estimate_data, sheet, all_date)
            
            # 计算距离上次买入的涨跌
            whetherNotToBuy(sheet)
        
        # 保存数据
        workbook.save('./基金数据买卖点.xlsx')


if __name__ == "__main__":
    # 基金代码,自行更改为关注的基金代码
    codes = ['161725', '005827', '003095','004788','014528','516160']
    # 获取表格对象
    workbook = load_workbook('./基金数据买卖点.xlsx')
    # 调用函数
    invoking_func(codes,workbook)

    print('完成')

四、参考文章