18章
前回まで作ってきたクラスをまとめる作業。
今まで全部チェックフォルダにファイルを詰め込んでいたが、
本と同じ並びに変えた。
他のフォルダのモジュールをインポートするには、
import sys,os sys.path.append(os.pardir + "\lib")
を最初に追加することでできるようになる。
ただし以前のコードでDataフォルダを参照するときに
os.pardirを追加する必要が出てくるようになった。
他のコードもエラー処理でちょこちょこ変更したりしているので、
すべてが終わった時に修正するつもり。
libフォルダのtrading_system.py
rubyのflattenがないので_flattenメソッドを追加している。
(参考)pythonでflatten
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# トレーディングシステムの各ルールを管理するクラス
class TradingSystem:
def __init__(self, rules = {}):
self.entries = [x for x in list(self._flatten(rules['entries'])) if x is not None]
self.exits = [x for x in list(self._flatten(rules['exits'])) if x is not None]
self.stops = [x for x in list(self._flatten(rules['stops'])) if x is not None]
self.filters = [x for x in list(self._flatten(rules['filters'])) if x is not None]
def set_stock(self, stock):
def func(rule):
rule.stock = stock
return self._each_rules(func)
def calculate_indicators(self):
def func(rule):
rule.calculate_indicators()
return self._each_rules(func)
# フィルターを適用して仕掛けをチェックする
def check_entry(self, index):
trade = self._entry_through_filter(index)
if trade is not None:
return self._trade_with_first_stop(trade, index)
# ストップを設定する
def set_stop(self, position, index):
position.stop = self._tightest_stop(position, index)
# 各手仕舞いルールを順にチェックし、
# 最初に手仕舞いが発生した時点で手仕舞う
# 中には、手仕舞いを制限するルールもある
def check_exit(self, trade, index):
for exit_rule in self.exits:
exit_filter = exit_rule.check_exit(trade, index)
if exit_filter == 'no_exit':
return None
if trade.closed_check == True:
return None
def _each_rules(self, func):
for rule in list(self._flatten([self.entries, self.exits, self.stops, self.filters])):
func(rule)
def _entry_through_filter(self, index):
if self._filter_signal(index) == 'no_entry':
return None
elif self._filter_signal(index) == 'long_and_short':
return self._check_long_entry(index) or self._check_short_entry(index)
elif self._filter_signal(index) == 'long_only':
return self._check_long_entry(index)
elif self._filter_signal(index) == 'short_only':
return self._check_short_entry(index)
# すべてのフィルターをチェックして、
# 仕掛けられる条件を絞る
def _filter_signal(self, index):
filters = [my_filter.get_filter(index) for my_filter in self.filters]
if None in filters or 'no_entry' in filters or ('long_only' in filters and 'short_only' in filters):
return 'no_entry'
elif 'long_only' in filters:
return 'long_only'
elif 'short_only' in filters:
return 'short_only'
else:
return 'long_and_short'
# 各仕掛けルールを順にチェックし、
# 最初に買い仕掛けが生じた時点で新規買トレードを返す
# 仕掛けがなければnilを返す
def _check_long_entry(self, index):
return self._check_entry_rule('long', index)
# 各仕掛けルールを順にチェックし、
# 最初に売り仕掛けが生じた時点で新規売トレードを返す
# 仕掛けがなければnilを返す
def _check_short_entry(self, index):
return self._check_entry_rule('short', index)
def _check_entry_rule(self, long_short, index):
for entry in self.entries:
trade = getattr(entry, 'check_{0}_entry'.format(long_short))(index)
if trade is not None:
return trade
return None
# 最もきついストップの値段を求める
def _tightest_stop(self, position, index):
stops = [position.stop]
for stop in self.stops:
stops.append(stop.get_stop(position, index))
stops = [x for x in stops if x is not None]
if position.long_check:
return max(stops)
elif position.short_check:
return min(stops)
# 仕掛けの際の初期ストップを設定する
def _trade_with_first_stop(self, trade, index):
if len(self.stops) == 0:
return trade
stop = self._tightest_stop(trade, index)
# まだひとつもストップがなければ仕掛けない
if stop is None:
return None
trade.first_stop = stop
trade.stop = stop
return trade
def _flatten(self, i):
for a in i:
if hasattr(a, '__iter__'):
for b in self._flatten(a):
yield b
else:
yield acheckフォルダのtrading_system_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
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
trades = simulate(8604)
for trade in trades:
print(trade.entry_date)
print(trade.entry_price)
print(trade.entry_time)
print(trade.exit_date)
print(trade.exit_price)
print(trade.exit_time)
print(trade.length)
print(trade.stock_code)
print(trade.stop)
print(trade.trade_type)
print(trade.volume)
for trade in trades:
print(trade.profit())
print('総損益 ' + str(sum([trade.profit() for trade in trades])))
