ビクター・ニーダーホッファーのランチショー
少し前の話になるが、
ビクター・ニーダーホッファーの
ランチショー(講演会?)に参加してきた。
話の中身はその前日に行われた無料の講演会と
ほぼ同じという衝撃的なものだったが、
ビクター・ニーダーホッファーを近くで見られたということと、
参加者も高額な講演会だけあって、FXで有名な方などが来られていて、
そういった方々と話ができたので、
とても有意義な時間を過ごすことができた。
ビクター・ニーダーホッファーは、
天才とはこういうものなのかなという感じで、
悩みなんてものとは無縁な印象を受けた。
毎回だと困るが、
たまにはこういうセミナーがあってもいいね。
20章
すべてをまとめる部分。
チェック段階でやっと気づいたが、TextToStockクラスで、
'market_section'を指定しないとうまく動かなかった。
以前の作り方がまずかったらしい。
仕方なく今回は'market_section':'東証1部'と指定している。
さかのぼっての大幅な修正よりもとりあえずの完成を目指したい。
libフォルダのsimulation.py
#!/usr/bin/env python # -*- coding: utf-8 -*- from trading_system import TradingSystem from recorder import Recorder # 売買シミュレーションを行うクラス class Simulation: def __init__(self, params = {}): self.trading_system = params['trading_system'] self.data_loader = params['data_loader'] self.recorder = params['recorder'] self.frm = params['from'] self.to = params['to'] self.record_every_stock = False if params['record_every_stock'] == False else True # 1銘柄のシミュレーションを行う def simulate_a_stock(self, code): self._set_dates_to_data_loader() stock = self.data_loader.generate_stock(code) print(stock.code) self._simulate(stock) if len(self.trades) != 0: self.recorder.record_a_stock(self.trades) # すべての銘柄のシミュレーションを行う def simulate_all_stocks(self): results = [] self._set_dates_to_data_loader() for stock in self.data_loader.each_stock(): self._simulate(stock) print(stock.code) if len(self.trades) == 0: continue if self.record_every_stock: self.recorder.record_a_stock(self.trades) results.append(self.trades) self.recorder.record_stats_for_each_stock(results) self.recorder.record_stats(results) def _set_dates_to_data_loader(self): self.data_loader.frm = self.frm self.data_loader.to = self.to def _simulate(self, stock): self.trading_system.set_stock(stock) self.trading_system.calculate_indicators() self.trades = [] self.position = None self.unit = stock.unit for index in range(0, len(stock.prices)): self.index = index self._before_open() self._at_open() self._in_session() self._at_close() def _before_open(self): self.signal = None if self.position is not None: self.position.exit_date = None self.position.exit_price = None self.trading_system.set_stop(self.position, self.index) self.position.length += 1 def _at_open(self): self._take_position('open') self._close_position('open') def _in_session(self): self._take_position('in_session') self._close_position('in_session') def _at_close(self): self._take_position('close') self._close_position('close') def _take_position(self, entry_time): if self.position is None: if self.signal is None: self.signal = self.trading_system.check_entry(self.index) if self.signal is not None: if self.signal.entry_time == entry_time: self.position = self.signal self.position.volume = self.unit self.signal = None def _close_position(self, exit_time): if self.position is not None: if self.position.closed_check() == False: self.trading_system.check_exit(self.position, self.index) if self.position.closed_check() == True: if self.position.exit_time == exit_time: self.trades.append(self.position) self.position = None
checkフォルダのsimulation_check.py
#!/usr/bin/env python # -*- coding: utf-8 -*- import sys, os sys.path.append(os.pardir + "\lib") from simulation import Simulation from text_to_stock import TextToStock from trading_system import TradingSystem from recorder import Recorder 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 text_to_stock = TextToStock({'data_dir': os.pardir + '\\data', 'stock_list':'tosho_list.txt', 'market_section':'東証1部'}) estrangement_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})]}) recorder = Recorder() recorder.record_dir = os.pardir + r'\result\estrangement\test_simulation' simulation = Simulation({'trading_system' : estrangement_system, 'data_loader' : text_to_stock, 'recorder' : recorder, 'from' : None, 'to' : None, 'record_every_stock': False}) recorder.create_record_folder() simulation.simulate_a_stock(8604) simulation.simulate_all_stocks()
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.percentages
libフォルダの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)
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 a
checkフォルダの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])))
17章
売買ルールつくり。
とりあえず作り終え、チェックファイルも作ったが、エラーが出た。
rubyのyieldは匿名関数に近いということなので、
rule.pyの_with_valid_indicatorsの引数に式を取るようにしたのだが、
それがどうもまずく、self.check_long(index)の評価の段階で
__getitem__のチェックが入り、Noneなので
ValueErrorが出てしまい止まってしまう。
結局entry.pyのcheck_long_entryの部分を以下のように変えることで動くようになった。
他のファイルはまだチェックしてないが同じようにしないとダメだろう。
def check_long_entry(self, index): try: return self._with_valid_indicators(self.check_long(index)) except ValueError: pass def check_short_entry(self, index): try: return self._with_valid_indicators(self.check_short(index)) except ValueError: pass
以下順番に
estrangement_entry.py
estrangement_exit.py
stop_out_exit.py
average_true_range_stop.py
moving_average_direction_filter.py
estrangement_entry_check.py
#!/usr/bin/env python # -*- coding: utf-8 -*- from entry import Entry from estrangement import Estrangement # 移動平均乖離率による仕掛けクラス # 前日の終値が、n日移動平均からx%離れたら寄付きで仕掛ける class EstrangementEntry(Entry): def __init__(self, params): self.span = params['span'] self.rate = params['rate'] def calculate_indicators(self): self.estrangement = Estrangement(self.stock, {'span': self.span}).calculate() return self.estrangement def check_long(self, index): if self.estrangement[index - 1] < (-1) * self.rate: return self._enter_long(index, self.stock.open_prices()[index], 'open') def check_short(self, index): if self.estrangement[index - 1] > self.rate: return self._enter_short(index, self.stock.open_prices()[index], 'open')
#!/usr/bin/env python # -*- coding: utf-8 -*- from exit import Exit from estrangement import Estrangement # 移動平均乖離率による手仕舞いクラス # 前日の終値が、n日移動平均からx%以内のとき寄付きで手仕舞う class EstrangementExit(Exit): def __init__(self, params): self.span = params['span'] self.rate = params['rate'] def calculate_indicators(self): self.estrangement = Estrangement(self.stock, {'span': self.span}).calculate() def check_long(self, trade, index): if self.estrangement[index - 1] > (-1) * self.rate: self._exit(trade, index, self.stock.open_prices()[index], 'open') def check_short(self, trade, index): if self.estrangement[index - 1] < self.rate: self._exit(trade, index, self.stock.open_prices()[index], 'open')
#!/usr/bin/env python # -*- coding: utf-8 -*- from exit import Exit # ストップアウトクラス # 価格がストップに掛かったら手仕舞う # ストップに掛かった瞬間、即座に手仕舞う # 寄付きで掛かったら寄付きで、 # 場中に掛かったら場中に手仕舞う class StopOutExit(Exit): def check_long(self, trade, index): stop = trade.stop price = self.stock.prices[index] if stop >= price['low']: if stop >= price['open']: price, time = price['open'], 'open' else: price, time = stop, 'in_session' self._exit(trade, index, price, time) def check_short(self, trade, index): stop = trade.stop price = self.stock.prices[index] if stop <= price['high']: if stop <= price['open']: price, time = price['open'], 'open' else: price, time = stop, 'in_session' self._exit(trade, index, price, time)
#!/usr/bin/env python # -*- coding: utf-8 -*- from stop import Stop from average_true_range import AverageTrueRange import Tick # 真の値幅の移動平均(ATR)に基づくストップクラス # 仕掛値から # 買いでは、n日間のATRのx倍、下に # 売りでは、n日間のATRのx倍、上に # ストップを置く class AverageTrueRangeStop(Stop): # span ATRの期間 # ratio ATRの何倍か def __init__(self, params): self.span = params['span'] self.ratio = params['ratio'] or 1 def calculate_indicators(self): self.average_true_range = AverageTrueRange(self.stock, {'span': self.span}).calculate() # 仕掛値からn日ATR(前日)のx倍下に行ったところにストップ def stop_price_long(self, position, index): Tick.truncate(position.entry_price - self._range(index)) # 仕掛値からn日ATR(前日)のx倍上に行ったところにストップ def stocp_price_short(self, position, index): Tick.ceil(position.entry_price + self._range(index)) def _range(self, index): self.average_true_range[index - 1] * self.ratio
#!/usr/bin/env python # -*- coding: utf-8 -*- from filter import Filter from moving_average_direction import MovingAverageDirection # 移動平均の方向フィルタークラス # 前日の移動平均が上昇中のときは、買いからのみ入る # 前日の移動平均が下降中のときは、売りからのみ入る # 前日の移動平均が横ばいのときは、仕掛けない class MovingAverageDirectionFilter(Filter): def __init__(self, params): self.span = params['span'] def calculate_indicators(self): self.moving_average_direction = MovingAverageDirection(self.stock, {'span': self.span}).calculate() def filter(self, index): if self.moving_average_direction[index - 1] == 'up': return 'long_only' elif self.moving_average_direction[index - 1] == 'down': return 'short_only' elif self.moving_average_direction[index - 1] == 'flat': return 'no_entry'
#!/usr/bin/env python # -*- coding: utf-8 -*- from estrangement_entry import EstrangementEntry from text_to_stock import TextToStock tts = TextToStock({'data_dir':'data', 'stock_list':'tosho_list.txt', 'market_section':'東証1部'}) tts.frm = '2011/01/04' tts.to = '2011/06/30' stock = tts.generate_stock(8604) est_entry = EstrangementEntry({'span': 20, 'rate':5}) est_entry.stock = stock est_entry.calculate_indicators() real_signals = [None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,'long',None,None,None,None,None,None,None,None,None,None,'short','short',None,None,None,None,None,None,None,None,None,None,'long','long',None,'long','long','long','long','long','long','long',None,'long','long','long','long','long','long',None,'long',None,'long','long','long',None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,'long',None,None,None,None,'long','long',None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,None,'long','long','long',None,None,None,None,None,None,None] for i , sig in enumerate(real_signals): calc_signal = est_entry.check_long_entry(i+1) or est_entry.check_short_entry(i+1) if calc_signal is None: if real_signals[i] is None: print('None None') else: print('None ' + real_signals[i]) else: print(calc_signal.trade_type + ' ' + real_signals[i]) print(est_entry.check_long_entry(0)) print(est_entry.check_short_entry(0))