该部分记录如何获取预期结果-接口响应数据，分成两步：
1 获取数据源接口数据
2 提取后续页面对比中要用到的数据
并且为了便于后续调用，将接口相关的都封装到ProjectApi类中。
新建python包：apiclass 》 新建python file：api_fund。所有接口相关的操作均放到该文件中。隐去项目相关信息后的代码如下：
1 获取数据源接口数据
# coding:utf-8 import requests from common.round_rewrite import round_rewrite #点击查看该函数 四舍五入 from common.jsonp_to_json import jsonp_to_json #点击查看该函数 jsonp转成json class Fund: fund_api_url = '接口地址url' """四个原始数据接口""" def api_strategy(self,day=''): """ 接口1 """ url = Fund.fund_api_url+'api1.json' response = requests.get(url,params={"day":day}).json() return response def api_lastestinfo(self,code): """ 接口2 """ url = Fund.fund_api_url+'latestInfo/{0}.json'.format(code) response = requests.get(url).json() return response def api_trends(self,code,pattern,period): """ 接口3 """ identifier = "{code}_{pattern}_{period}".format(code=code,pattern=pattern,period=period) url = Fund.fund_api_url+"trends/{0}.json".format(identifier) jsonpstr = requests.get(url).text jsonstr = jsonp_to_json(jsonpstr) return jsonstr def api_timeline(self,code): """ 接口4 """ url = Fund.fund_api_url+"timeline/{0}.json".format(code) response = requests.get(url).json() return response
2 提取后续页面对比中要用到的数据
接口1比较特别，返回数据是一个list，按时间升序排列。有的页面需要取最早的数据，有的页面取最新的数据。
1 用一个参数来标识，该参数设置成list切片步长。1从前往后去，-1从后往前取。
2 samples是一个dict组成的list，要取出每一个dict里的values
samples = sorted([list(ele.values()) for ele in samples]) # 转变成与页面数据一致的格式
def get_fund_strategy(self,code,day='',latest=1): """提取接口1的数据 latest:1-取最早的数据；-1-取最新的数据""" fund_strategy = self.api_strategy(day) #获取策略配置接口数据，day之后的数据都会取出来 for ele in fund_strategy[::latest]: #1从前往后取最早，-1从后往前取最新 if ele["code"] == code: self.code = ele["code"] self.name = ele["name"] self.summary = ele["summary"] self.memo = ele["memo"] samples = ele["samples"] self.samples = sorted([list(ele.values()) for ele in samples]) # 转变成与页面数据一致的格式 return #取到则退出循环
接口2：一个页面只需要3M的数据，单独写了一个函数； 另一个页面需要全量数据。
def get_fund_latestinfo(self,code): """提取接口2的数据""" fund_lastestinfo = self.api_lastestinfo(code) nav = fund_lastestinfo["nav"] navDate = fund_lastestinfo["navDate"][-5:] navChange = fund_lastestinfo["navChange"] annualChangeAll = fund_lastestinfo["annualChangeAll"] self.navlist = [nav,navDate,navChange,annualChangeAll] percents = fund_lastestinfo["percents"] self.percents_list = [list(ele.values())[1:] for ele in percents]
def get_fund_percentM3(self,code): """获取3个月收益率，首页有该数据""" fund_lastestinfo = self.api_lastestinfo(code) self.percentM3 =fund_lastestinfo["percents"][0]["percentM3"] return self.percentM3
接口3：需要对数值进行判断，当数值>=0,显示超出，否则跑输。
sharprun = "超出" if self.sharpeDiff >= 0 else "跑输"
将列表里的每一个字典的key转成中文，方便与页面数据对比
self.trends = map(lambda line: {"日期":line["date"],"组合市值":line["mv"],"比较基准市值":line["bmv"]},trends) #列表里的字典key英文转成中文
def get_fund_trends(self,code,pattern,peroid): """提取接口3的数据""" fund_trends = self.api_trends(code, pattern, peroid) # 请求接口数据 """获取接口字段值""" self.percent = fund_trends["percent"] self.percentDiff = fund_trends["percentDiff"] self.maxDown = fund_trends["maxDown"] self.mdStart = fund_trends["mdStart"] self.mdEnd = fund_trends["mdEnd"] self.startDate = fund_trends["startDate"] try: self.sharpe = fund_trends["sharpe"] self.sharpeDiff = fund_trends["sharpeDiff"] sharprun = "超出" if self.sharpeDiff >= 0 else "跑输" percentRun = "超出" if self.percentDiff >= 0 else "跑输" sharpeDiff = abs(self.sharpeDiff) except KeyError: # 夏普比率跨年时今年以来接口无数据，置为空 sharpe = sharpeDiff = sharprun = percentRun = "" trends = fund_trends["trends"] # 组合涨幅走势数据 self.trends = map(lambda line: {"日期":line["date"],"组合市值":line["mv"],"比较基准市值":line["bmv"]},trends) #列表里的字典key英文转成中文 result = [code,self.startDate, percentRun, abs(self.percentDiff), self.percent, self.maxDown,sharprun, sharpeDiff, self.sharpe, self.mdStart, self.mdEnd] #与页面一样的格式 return result
接口4：需要取当前和昨天的值计算涨跌幅，并保留2位小数
"""提取原始接口中页面所需数据""" def get_fund_timeline(self,code): """提取接口4的数据""" fund_timeline = self.api_timeline(code) last = fund_timeline["last"] date = fund_timeline["date"] current = fund_timeline["current"] timeline = fund_timeline["timeline"] rate = (current - last) / last * 100 timeline_current = dict(日期=date,实时估值=current,估值涨幅=round_rewrite(rate,2)) timeline_list = [] for tl in timeline: #分时数据 dt = tl["dt"] nav = tl["nav"] rt = (nav - last) / last * 100 timelinedata = dict(时间=dt,估值=nav,涨跌幅=round_rewrite(rt,2)) timeline_list.append(timelinedata) return timeline_current,timeline_list
最后一部分，因页面图形无法自动化验证，手工测试相关的函数：
def mannualtest_timeline(self): """分时图手工测试""" print('code：00X00Y') scode = input("获取实时估值 输入code") try: """实时估值""" current, timeline = self.get_fund_timeline(scode) print("实时估值：", current) print("分时图数据：") for line in timeline: print(line) except Exception as e: print(e)
def mannualtest_trends(self): """走势图手工测试""" print('code：00X00Y') scode = input("获取组合走势图形数据：请输入code ") pattern = input("投资方式 W（默认） K ") peroiddict = {'1': 'R1M', '2': 'R3M', '3': 'R6M','5': 'R1Y', '6': 'R3Y'} peroid = input("投资期限输入对应数字 %s "%peroiddict) peroid = peroid.strip() pattern = pattern.strip().upper()#去除左右空格后转大写 if pattern != 'K': pattern = 'W'#只要不等于K，则默认W if peroid in peroiddict.keys(): peroid = peroiddict[peroid] #在字典里则取对应值 else: peroid = 'R1M' #不在字典取默认R1M try: self.get_fund_trends(scode, pattern, peroid) #获取接口数据 print("组合走势图{scode}_{pattern}_{peroid}".format(scode=scode,pattern=pattern,peroid=peroid)) for line in self.trends: print(line) except Exception as e: print(e)
the end！