19章
システムの成績を出す部分。
ensure_closeのメソッドを少し変更した。
libフォルダのstats.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from trade import Trade
from myarray import Array
# Statsクラス
# 取引結果から各種統計を計算
class Stats:
def __init__(self, trade):
self.trades = trade
def sum_profit(self):
return sum(self._profits())
def average_profit(self):
return (self._profits().average())
def wins(self):
return len([profit for profit in self._profits() if profit > 0])
def losses(self):
return len([profit for profit in self._profits() if profit > 0])
def draws(self):
return len([profit for profit in self._profits() if profit == 0])
def winning_percentage(self):
return self.wins() / len(self._profits())
def profit_factor(self):
if self.losses() == 0:
return None
else:
total_profit = sum([profit for profit in self._profits() if profit > 0])
total_loss = sum([profit for profit in self._profits() if profit < 0])
return total_profit / abs(total_loss)
def sum_r(self):
if any(self._r_multiples()):
return sum(self._r_multiples())
def average_r(self):
if any(self._r_multiples()):
return self._r_multiples().average()
def sum_percentage(self):
return sum(self._percentages())
def average_percentage(self):
return self._percentages().average()
def average_length(self):
return Array([trade.length for trade in self.trades]).average()
def _profits(self):
if hasattr(self, 'profits') == False:
self.profits = Array([trade.profit() for trade in self.trades])
return self.profits
def _r_multiples(self):
if hasattr(self, 'r_multiples') == False:
self.r_multiples = Array([trade.r_multiple() for trade in self.trades])
return self.r_multiples
def _percentages(self):
if hasattr(self, 'percentages') == False:
self.percentages = Array([trade.percentage_result() for trade in self.trades])
return self.percentageslibフォルダのrecorder.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
from trade import Trade
from myarray import Array
from stats import Stats
import re
import sys
import csv
import shutil
import os
import errno
from collections import OrderedDict
# 取引の記録を行うクラス
class Recorder:
def __init__(self, record_dir = None):
self.record_dir = record_dir
# 1銘柄の取引を記録する
def record_a_stock(self, trades):
code = trades[0].stock_code
file_name = '{0}/{1}.csv'.format(self.record_dir, code)
def func():
with open(file_name, 'w') as f:
writer = csv.writer(f, lineterminator='\n')
writer.writerow([value for value in self._items_for_a_stock().values()])
for trade in trades:
one_trade = []
for attr in self._items_for_a_stock().keys():
if callable(getattr(trade, attr)):
one_trade.append(getattr(trade, attr)())
else:
one_trade.append(getattr(trade, attr))
writer.writerow(one_trade)
self._ensure_close(file_name, func)
# 銘柄ごとの統計の一覧表の作成
def record_stats_for_each_stock(self, results):
file_name = '{0}/_stats_for_each_stock.csv'.format(self.record_dir)
def func():
with open(file_name, 'w') as f:
writer = csv.writer(f, lineterminator='\n')
line = [value for value in self._stats_items().values()]
line.insert(0, 'コード')
writer.writerow(line)
for trades in results:
line = self._stats_array(trades)
line.insert(0, trades[0].stock_code)
writer.writerow(line)
self._ensure_close(file_name, func)
# すべてのトレードの統計
def record_stats(self, results):
file_name = '{0}/_stats.csv'.format(self.record_dir)
def func():
with open(file_name, 'w') as f:
writer = csv.writer(f, lineterminator='\n')
writer.writerow([value for value in self._stats_items().values()])
writer.writerow(self._stats_array(list(self._flatten(results))))
self._ensure_close(file_name, func)
# 設定ファイルをコピーする
def record_setting(self, file_name):
shutil.copyfile(file_name, '{0}/_setting.py'.format(self.record_dir))
# 結果保存用のフォルダを作る
def create_record_folder(self):
if os.path.exists(self.record_dir):
print('記録フォルダ {0} はすでに存在します。上書きしますか? y/n'.format(self.record_dir))
def func():
print('上書きします')
self._yes_check(func)
else:
print('記録フォルダ {0} は存在しません。新しく作りますか? y/n'.format(self.record_dir))
def func():
os.mkdir(self.record_dir)
self._yes_check(func)
def _items_for_a_stock(self):
return OrderedDict((('trade_type' , '取引種別'),
('entry_date' , '入日付'),
('entry_price' , '入値'),
('volume' , '数量'),
('first_stop' , '初期ストップ'),
('exit_date' , '出日付'),
('exit_price' , '出値'),
('profit' , '損益(円)'),
('r_multiple' , 'R倍数'),
('percentage_result', '%損益'),
('length' , '期間')))
def _stats_items(self):
return OrderedDict((('sum_profit' , '総損益'),
('wins' , '勝ち数'),
('losses' , '負け数'),
('draws' , '分け数'),
('winning_percentage', '勝率'),
('average_profit' , '平均損益'),
('profit_factor' , 'PF'),
('sum_r' , '総R倍数'),
('average_r' , '平均R倍数'),
('sum_percentage' , '総損益率'),
('average_percentage', '平均損益率'),
('average_length' , '平均期間')))
def _stats_array(self, trades):
sa = Array([])
stats = Stats(trades)
for stats_name in self._stats_items().keys():
sa.append(getattr(stats, stats_name)() or '-')
return sa
def _ensure_close(self, file_name, func):
try:
func()
except PermissionError as e:
if e.errno == errno.EACCES:
while True:
i = input('{0}が他のプログラムで書き込み禁止で開かれている可能性があります。\nファイルを閉じてからエンターキーを押してください。'.format(file_name))
if not i:
func()
break
def _yes_check(self, func):
while True:
i = input('')
if re.search('^[yY]', i) is not None:
func()
break
elif re.search('^[nN]', i) is not None:
print('終了します')
sys.exit()
else:
print('y(はい) か n(いいえ)でお答えください')
def _flatten(self, i):
for a in i:
if hasattr(a, '__iter__'):
for b in self._flatten(a):
yield b
else:
yield a
checkフォルダのrecorder_check.py
#!/usr/bin/env python
# -*- coding: utf-8 -*-
import sys, os
sys.path.append(os.pardir + "\lib")
from trading_system import TradingSystem
from text_to_stock import TextToStock
from rule.entry.estrangement_entry import EstrangementEntry
from rule.exit.stop_out_exit import StopOutExit
from rule.exit.estrangement_exit import EstrangementExit
from rule.stop.average_true_range_stop import AverageTrueRangeStop
from rule.filter.moving_average_direction_filter import MovingAverageDirectionFilter
from recorder import Recorder
data = TextToStock({'data_dir': None, 'stock_list': 'tosho_list.txt', 'market_section': None})
trading_system = TradingSystem({'entries': [EstrangementEntry({'span': 20, 'rate': 5})],
'exits' : [StopOutExit(),
EstrangementExit({'span': 20, 'rate': 3})],
'stops' : [AverageTrueRangeStop({'span': 20, 'ratio': None})],
'filters': [MovingAverageDirectionFilter({'span': 30})]})
def simulate(code):
stock = data.generate_stock(code)
trading_system.set_stock(stock)
trading_system.calculate_indicators()
trade = None
trades = []
for i in range(0,len(stock.prices)):
if trade is not None:
trading_system.set_stop(trade, i)
trade.length += 1
if trade is None:
trade = trading_system.check_entry(i)
if trade is not None:
trade.volume = stock.unit
if trade is not None:
trading_system.check_exit(trade, i)
if trade.closed_check() == True:
trades.append(trade)
trade = None
return trades
recorder = Recorder()
recorder.record_dir = os.pardir + r'\result\test'
recorder.create_record_folder()
recorder.record_setting(__file__)
results = [simulate(code) for code in [4063, 7203, 8604]]
results = [trades for trades in results if len(trades) != 0]
for trades in results:
recorder.record_a_stock(trades)
recorder.record_stats_for_each_stock(results)
recorder.record_stats(results)
