1. Johngo学长首页
  2. 技术点详细复盘专题
  3. Python

(量化) 用 Tushare 包实现一个简单的回测框架

• Python • 阅读 41

【Reference】
1. B站：清华计算机博士带你学-Python金融量化分析
2. Tushare 官网（作者ID：492952）

本博客所包含的项目代码基本参考Reference1，并对其中tushare API的更新做了对应的修改。全部代码已上传至作者的 github，下文中仅针对代码思路做一个知识梳理

1. 回测需求 & 效果展示

以中国平安（601318.SH）为买卖对象，检验 双均线策略在 2020-05-10 至 2021-01-01 的收益情况。效果如下：

(量化) 用 Tushare 包实现一个简单的回测框架

; 2. 代码框架

2.1 对象

(1) 存储账户信息、回测信息

账户信息：现金、股票持有量

[En]

Account information: cash, stock holdings

回测信息：

  1. 开始/结束日期
  2. 当前日期
  3. 基准：一般会以一只股票或一个指数为基准，用于比较策略优劣性
  4. 开始至结束之间所有交易日的信息
class Context:
    def __init__(self, cash, start_date, end_date):

        self.cash = cash
        self.positions = {}

        self.start_date = start_date
        self.end_date = end_date
        self.dt = start_date
        self.benchmark = None
        self.date_range = trade_cal[(trade_cal['is_open'] == '1') & \
                                    (trade_cal['cal_date'] >= start_date) & \
                                    (trade_cal['cal_date']  end_date)]['cal_date'].values

其中， trade_cal 存储了所有交易日的日期信息。可通过 tushare 包的 pro.trade_cal() 获取

(2) 存储其他全局变量

class G:
    pass

2.2 函数

(0) 主体函数

def run():
    initialize(context)

    plt_df = pd.DataFrame(index=pd.to_datetime(context.date_range), columns=['value'])

    last_prices = {}

    initial_value = context.cash

    for dt in context.date_range:
        context.dt = dt
        handle_data(context)

        value = context.cash
        for stock in context.positions.keys():
            try:
                data = get_today_data(stock)
                last_prices[stock] = data['open']
            except KeyError:

                price = last_prices[stock]
            value += price * context.positions[stock].amount
        plt_df.loc[dt, 'value'] = value

    plt_df['ratio'] = (plt_df['value'] - initial_value) / initial_value

    bm_df = attribute_daterange_history(context.benchmark, context.start_date, context.end_date)
    bm_init = bm_df['open'][0]
    plt_df['benchmark_ratio'] = (bm_df['open'] - bm_init) / bm_init

    plt_df[['ratio', 'benchmark_ratio']].plot()
    plt.show()

(1) 初始化函数

  1. 设置目标股票
  2. 设置基准
  3. 设置双均线信息（g.p1 & g.p2）
  4. 获取目标股票的历史数据：在以下代码中， hist_1 获取的是回测开始日期前 g.p2 天的数据， hist_2 获取的是回测开始至结束之间的数据。将两者合并后即可得到双均线回测所需要的所有股票价格数据
def initialize(context):
    g.security = '601318.SH'
    set_benchmark('601318.SH')
    g.p1 = 5
    g.p2 = 60

    hist_1 = attribute_history(g.security, g.p2)
    hist_2 = attribute_daterange_history(g.security, context.start_date, context.end_date)
    g.hist = hist_1.append(hist_2)

(2) 每个交易日都需执行的函数

def handle_data(context):
    hist = g.hist[:dateutil.parser.parse(context.dt)][-g.p2:]
    ma5 = hist['close'][-g.p1:].mean()
    ma60 = hist['close'].mean()

    if ma5 > ma60 and g.security not in context.positions.keys():
        order_value(g.security, context.cash)
    elif ma5 < ma60 and g.security in context.positions.keys():
        order_target(g.security, 0)

(3) 下单函数

github 中上传的代码共包含四个下单函数:

  1. order: 购买一定的股票数量
  2. order_value 购买一定的股票金额
  3. order_target 购买到一定数量的股票
  4. order_target_value 购买到一定金额的股票

这四个下单函数都是基于：

[En]

These four order placing functions are all based on:

def _order(today_data, security, amount):
    if today_data.empty: return

    price = today_data['open']

    try:
        test = context.positions[security]
    except KeyError:

        if amount  0: return

        context.positions[security] = pd.Series(dtype=float)
        context.positions[security]['amount'] = 0

    if (amount % 100 != 0) and (amount != -context.positions[security].amount):
        amount = int(amount/100) * 100

    context.positions[security].amount = context.positions[security].get('amount') + amount
    if context.positions[security].amount == 0:
        del context.positions[security]

    context.cash -= amount * price

Original: https://blog.csdn.net/weixin_43728138/article/details/122859016
Author: 江湖留名
Title: (量化) 用 Tushare 包实现一个简单的回测框架

相关阅读

Title: 国内访问Github超级慢？那是你没有用我这个脚本。直接起飞。

导语

之前很多朋友咨询过国内访问Github较慢的问题，然后我一般让他们自己去知乎上找攻略，但今天我才发现网上竟然没有一个一键配置的脚本，一般都需要我们跟着教程一步步地去做才行。这也太麻烦了，于是自己动手写了个脚本，只需要简单运行一下就可以实现访问Github加速的功能。

废话不多说，让我们愉快地开始吧~

开发工具

Python 版本：3.7.8

相关模块：

pikachupytools模块；
pythonping模块；
beautifulsoup4模块；
requests模块；
tqdm模块；
以及一些python自带的模块。

环境搭建

安装Python并添加到环境变量，pip安装需要的相关模块即可。

原理简介

核心原理就是把Github相关的域名和对应的ip地址写到本地hosts文件里，从而绕过DNS解析，达到加速访问的目的。

其中Github相关的域名整理如下：


domains = [
    'github.com', 'www.github.com', 'github.global.ssl.fastly.net', 'github.map.fastly.net', 'github.githubassets.com',
    'github.io', 'assets-cdn.github.com', 'gist.github.com', 'help.github.com', 'api.github.com', 'nodeload.github.com',
    'codeload.github.com', 'raw.github.com', 'documentcloud.github.com', 'status.github.com', 'training.github.com',
    'raw.githubusercontent.com', 'gist.githubusercontent.com', 'cloud.githubusercontent.com', 'camo.githubusercontent.com',
    'avatars0.githubusercontent.com', 'avatars1.githubusercontent.com', 'avatars2.githubusercontent.com', 'avatars3.githubusercontent.com',
    'avatars4.githubusercontent.com', 'avatars5.githubusercontent.com', 'avatars6.githubusercontent.com', 'avatars7.githubusercontent.com',
    'avatars8.githubusercontent.com', 'user-images.githubusercontent.com', 'favicons.githubusercontent.com', 'github-cloud.s3.amazonaws.com',
    'github-production-release-asset-2e65be.s3.amazonaws.com', 'github-production-user-asset-6210df.s3.amazonaws.com',
    'github-production-repository-file-5c1aeb.s3.amazonaws.com', 'alive.github.com', 'guides.github.com', 'docs.github.com'
]

然后我们利用如下网站查询每个域名对应的IP地址：

https://ipaddress.com/website/ + &#x57DF;&#x540D;

效果大概是这样子的：

(量化) 用 Tushare 包实现一个简单的回测框架

代码实现如下：

# 生成符合ipadredd.com查询的url地址, 从而解析出域名对应的IP
pbar = tqdm(self.domains)
for domain in pbar:
    pbar.set_description(f'parse {domain}')
    url = f'https://ipaddress.com/website/{domain}'
    response = self.session.get(url, headers=self.headers)
    soup = BeautifulSoup(response.text, 'lxml')
    ips = []
    for item in soup.find('tbody', id='dnsinfo').select('tr td a'):
        ip = re.findall(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", item.text)
        if ip: ips.append(''.join(ip))
    assert len(ips) > 0, f'parse {domain} error'
    if len(ips) == 1:
        domain2ip_dict[domain] = ''.join(ips)
    else:
        domain2ip_dict[domain] = self.lowestrttip(ips)

最后，我们根据操作系统的不同更新一下对应位置的hosts文件就行：


# 更新host文件
print(f'[INFO]: start to update the host file')
shutil.copy(self.hosts_path, self.hosts_path + '.bak')
fp_host, fp_temp = open(self.hosts_path, 'r'), open('temphost', 'w')
def distinct(domains, line):
    for domain in domains:
        if domain in line: return True
    return False
for line in fp_host.readlines():
    if not distinct(self.domains, line):
        fp_temp.write(line)
for domain, ip in domain2ip_dict.items():
    fp_temp.write(f'{domain}\t{ip}\n')
fp_host.close()
fp_temp.close()
shutil.copy('./temphost', self.hosts_path)
os.remove('./temphost')
if 'Windows' in platform.platform():
    os.system('ipconfig /flushdns')
else:
    os.system('systemd-resolve --flush-caches')
print(f'[INFO]: update the host file successfully')

更新之后的hosts文件大概长这样：

(量化) 用 Tushare 包实现一个简单的回测框架

大功告成啦，完整源代码详见相关文件~

效果展示

想要快速测试的小伙伴只需要pip安装一下pikachupytools包：

pip&#xA0;install&#xA0;pikachupytools&#xA0;--upgrade

然后简单写几行代码就能调用运行了：

from pytools import pytools

tool_client = pytools.pytools()
tool_client.execute('githubacceleration')

效果如下(必须管理员权限运行)：

(量化) 用 Tushare 包实现一个简单的回测框架

到这里就完美结束了，你学会了吗？源码见下方公众号！

Original: https://blog.csdn.net/weixin_43881394/article/details/127841935
Author: 爬遍天下无敌手
Title: 国内访问Github超级慢？那是你没有用我这个脚本。直接起飞。

原创文章受到原创版权保护。转载请注明出处：https://www.johngo689.com/296433/

转载文章受原作者版权保护。转载请注明原作者出处！

赞 (0)
0

【自取】最近整理的，有需要可以领取学习：



Linux核心资料大放送~

全栈面试题汇总（持续更新&可下载）

一个提高学习100%效率的工具！

【超详细】深度学习面试题目！

LeetCode Python刷题答案下载！

LeetCode Java版刷题答案下载！

LeetCode C++ 版本，抓紧保存！

LeetCode GO语言 刷题答案下载！

大家都在看

亲爱的 Coder【最近整理，可免费获取】👉 最新必读书单  | 👏 面试题下载