「米国株売買シミュレーションで学ぶPythonプログラミング」の4回目では、コマンドを実行できるところまで解説していきます。コマンド名は「dummy」で、コマンド自体はログを出力だけの単純な処理になります。
※当連載で扱うアプリのソースはGitHub上に公開してあります。 https://github.com/yusukemurayama/ppytrading
manager.py
まずは、コマンドを実行するときに叩くmanager.pyを作成します。manager.pyでは、main関数に処理の大半を記述していきます。
ソース全体
#!/usr/bin/env python
# coding: utf-8
import os
import sys
import argparse
import importlib
from ppyt.const import BASE_DIR
# locale.setlocale(locale.LC_ALL, 'en_US')
SCRIPT_DIR = os.path.join(BASE_DIR, 'commands')
def main():
# パスを追加します。
sys.path.append(BASE_DIR)
# 実行するコマンドを決定します。
if len(sys.argv) <= 1:
show_command_list() # コマンド一覧を表示して終了します。
# 実行するコマンド名と、そのコマンド名を除外した引数のリストを取得します。
command, newargs = None, []
parser = argparse.ArgumentParser()
parser.add_argument('command', type=str, nargs=1) # 最初のコマンドライン引数をコマンド名とします。
for args in parser.parse_known_args(): # parse_known_argsで余分な引数を無視します。
if hasattr(args, 'command'):
# nargs=1を指定しているので、サイズ1のlistにコマンド名が入っています。
command = args.command[0]
continue
newargs.extend(args) # コマンド名以外を引数のリストに追加していきます。
# 実行するスクリプトを決定します。
if command is None or command + '.py' not in os.listdir(SCRIPT_DIR):
show_command_list() # コマンド一覧を表示して終了します。
script_path = 'ppyt.commands.' + command
# 実行するクラスのインスタンスを生成してキックします。
module = importlib.import_module(script_path)
try:
instance = getattr(module, 'Command')(manager=os.path.basename(__file__),
command=command, args=newargs)
except Exception as e:
print(e)
print('コマンドを実行できませんでした。'
'{}がCommandBaseを継承しているかを確認してください。'.format(module.__name__))
exit()
instance.start()
def show_command_list():
print('# コマンド一覧')
for filename in os.listdir(SCRIPT_DIR):
command, ext = os.path.splitext(filename)
if ext != '.py' or command.startswith('_'):
continue
print('* ' + command)
exit()
if __name__ == '__main__':
main()
ソース解説
def main():
# パスを追加します。
sys.path.append(BASE_DIR)
ppytディレクトリをパスに追加しています。これにより、ppyt以下に配置したファイルで別モジュールをインポートするときに、from ppyt.models.orm import Stockのように、ppytからのパスを指定できるようになります。
# 実行するコマンドを決定します。
if len(sys.argv) <= 1:
show_command_list() # コマンド一覧を表示して終了します。
引数に実行するコマンドが含まれるかをチェックしています。なお、sys.argvの1番目は実行するスクリプト名が格納されているので、len(sys.argv) == 1のときは引数不足と判定しています。
# 実行するコマンド名と、そのコマンド名を除外した引数のリストを取得します。
command, newargs = None, []
...
newargs.extend(args) # コマンド名以外を引数のリストに追加していきます。
ここは、第2回でやったargparseを使ってコマンド名と、コマンド名以外の引数に分けています。
# 実行するスクリプトを決定します。
if command is None or command + '.py' not in os.listdir(SCRIPT_DIR):
show_command_list() # コマンド一覧を表示して終了します。
コマンドを置くディレクトリに、実行対象の「コマンド名.py」という名前のファイルが存在するかをチェックしています。os.listdirメソッドは、引数で指定したディレクトリ内のファイル名の一覧を返します。
# 実行するクラスのインスタンスを生成してキックします。
module = importlib.import_module(script_path)
importlibのimport_moduleメソッドを使ってPythonファイルを読み込みます。これは、例えばdummyコマンドの場合、from ppyt.commands import dummy as moduleと同じようなことを実行しています。なぜ「from...」ではなくimportlibを使ったかというと、インポートするモジュール名がコマンドごとに違うからです。dummyコマンド実行時はdummy.pyを読み込み、backtestコマンド実行時はbacktest.pyを読み込むことになります。
try:
instance = getattr(module, 'Command')(manager=os.path.basename(__file__),
command=command, args=newargs)
except Exception as e:
print(e)
print('コマンドを実行できませんでした。'
'{}がCommandBaseを継承しているかを確認してください。'.format(module.__name__))
exit()
instance.start()
モジュールを読み込んだら、getattr(module, 'Command')で、moduleの中で定義されているCommandクラスを取得できます。Commandクラスを取得したら「クラス名()」のようにインスタンスを生成します。abstract methodの実装漏れなどインスタンスの生成に失敗したら、exceptの中に入って原因を表示して終了します。
ppyt/commands/__init__.py
コマンドの基底クラスを作成します。すべてのコマンドは__init__.pyのCommandBaseクラスを継承して作成することになります。
ソース全体
※今回の範囲に関係ない部分は省いています。
# coding: utf-8
import logging
import time
import abc
import numpy as np
from ppyt.exceptions import CommandError
logger = logging.getLogger(__name__)
plogger = logging.getLogger('print')
class CommandBase(metaclass=abc.ABCMeta):
"""各種コマンドの規定クラスです。"""
def __init__(self, manager, command, args):
"""コンストラクタ
Args:
manager: managerスクリプトのファイル名
command: 実行するコマンド名
args: 引数のlist（コマンド名は除外されています。）
"""
self._manager = manager
self._command = command
self._args = args
@abc.abstractmethod
def _add_options(self, parser):
"""argparseのルールを記述します。サブクラスでオーバーライドします。"""
pass
@abc.abstractmethod
def _execute(self, options):
"""コマンドの本体です。サブクラスでオーバーライドします。"""
pass
def start(self):
start_time = time.time()
# 引数を解析します。
import sys
import argparse
parser = argparse.ArgumentParser(prog='{} {}'.format(sys.argv[0], self._command))
parser.add_argument('-v', '--verbose', action='store_true', default=False)
self._add_options(parser)
options = parser.parse_args(self._args)
self._set_logger(options.verbose)
np.seterr(invalid='ignore')
try:
logger.info('{}コマンドを開始します。'.format(self._command))
self._execute(options) # コマンドを実行します。
logger.info('{}コマンドが正常に終了しました。実行時間: {}秒'.format(
self._command, round(time.time()-start_time, 2)))
except CommandError as e:
# コマンド実行中に（軽微な）例外が発生したら画面に表示します。
plogger.info(e)
except Exception as e:
# コマンド実行中に例外が発生したらログに出力し、呼び出し元に投げます。
logger.error('{}コマンドが異常終了しました。原因: {}'.format(self._command, e))
raise e
def _set_logger(self, verbose):
"""ログ周りの設定をします。
Arguments:
verbose: Trueにすると詳細表示モードになります。
"""
level = logging.DEBUG if verbose else logging.INFO
fmt = '%(asctime)s - %(filename)s.%(funcName)s#%(lineno)s - %(levelname)s - %(message)s'
kwds = {}
if level == logging.INFO: # INFOレベルの場合は出力先をファイルにします。
# kwds['filename'] = const.LOG_FILEPATH
pass
logging.basicConfig(level=level, format=fmt, **kwds)
# SQLAlchemyのログレベルを変更します。
logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG if verbose else logging.WARN)
# メッセージ表示用のロガーを準備します。
plogger = logging.getLogger('print')
plogger.setLevel(logging.INFO)
ch = logging.StreamHandler()
ch.setLevel(logging.INFO)
formatter = logging.Formatter('%(message)s')
ch.setFormatter(formatter)
plogger.addHandler(ch)
plogger.propagate = False
ソース解説
startメソッド
manager.pyがstartメソッドを呼びます。このメソッドで初期処理などを行い、_executeメソッドを呼びます。
# 引数を解析します。
import sys
import argparse
parser = argparse.ArgumentParser(prog='{} {}'.format(sys.argv[0], self._command))
parser.add_argument('-v', '--verbose', action='store_true', default=False)
self._add_options(parser)
options = parser.parse_args(self._args)
argparseを使って引数を解析します。すべてのコマンドで「--verbose」を使えるようにして、後のパターンは各コマンドの_add_optionsメソッドで追加していくことになります。
np.seterr(invalid='ignore')
NumPyの、特定のエラーメッセージの表示を抑制しています。抑制するメッセージは、NaNが含まれる配列をnumpy.greaterメソッドで比較した場合などに出力されます。
import numpy as np
arr1 = np.array([1, 4, 3], dtype=np.float64)
arr2 = np.array([np.nan, 2, 9], dtype=np.float64)
np.greater(arr1, arr2) # RuntimeWarning: invalid value encountered in greater print(np.greater(arr1, arr2))
try:
logger.info('{}コマンドを開始します。'.format(self._command))
self._execute(options) # コマンドを実行します。
logger.info('{}コマンドが正常に終了しました。実行時間: {}秒'.format(
self._command, round(time.time()-start_time, 2)))
except CommandError as e:
# コマンド実行中に（軽微な）例外が発生したら画面に表示します。
plogger.info(e)
except Exception as e:
# コマンド実行中に例外が発生したらログに出力し、呼び出し元に投げます。
logger.error('{}コマンドが異常終了しました。原因: {}'.format(self._command, e))
raise e
_executeメソッド（コマンド本体）を実行しています。_executeメソッドで発生した例外は、CommandErrorとそれ以外に分けて処理しています。
set_loggerメソッド
set_loggerメソッドでは、ログ周りの初期設定を行っています。コマンド実行時の引数に「-v」か「--verbose」がある場合はDEBUGレベル以上、ない場合はINFOレベル以上のログをコンソール上に出力しています。
「plogger」は、画面にメッセージだけを表示するロガーです。print関数を使ってもいいのですが、loggingを使った方が出力先の変更など柔軟に対応できるので用意しています。
propagateについて
ploggerではpropagateにFalseを指定しています。このpropagateをFalseにしておくと、親（上位）への伝搬を防ぐことができます。これを設定しておかないと、plogger.info('FOO')のようなときに、「FOO」メッセージが重複して表示されてしまいます。
FOO # ploggerが出力
2016-xx-xx xx:xx:xx,xxx - dummy.py._execute#20 - INFO - FOO # 親（basicConfigで設定）が出力
propagateの例
下記コードでは、foobar1_loggerのみch1（foo_loggerで設定したハンドラ）とch2にログが出力されます。fooobar_loggerはpropagate=Trueにしていますが、foo_loggerの子供ではない（oの数が1つ多い）のでch1にはログが出力されません。
# coding: utf-8
import logging
fmt = '%(name)s - %(message)s'
ch1 = logging.StreamHandler()
ch1.setLevel(logging.DEBUG)
ch1.setFormatter(logging.Formatter('CH1: ' + fmt))
ch2 = logging.StreamHandler()
ch2.setLevel(logging.DEBUG)
ch2.setFormatter(logging.Formatter('CH2: ' + fmt))
foo_logger = logging.getLogger('foo')
foo_logger.setLevel(logging.DEBUG)
foo_logger.addHandler(ch1)
print('--- foo_logger ---')
foo_logger.info('FOO')
foobar1_logger = logging.getLogger('foo.bar1')
foobar1_logger.addHandler(ch2)
print('--- foobar1_logger (propagate: True) ---')
foobar1_logger.info('FOOBAR1')
foobar2_logger = logging.getLogger('foo.bar2')
foobar2_logger.addHandler(ch2)
foobar2_logger.propagate = False
print('--- foobar2_logger (propagate: False) ---')
foobar2_logger.info('FOOBAR2')
fooobar_logger = logging.getLogger('fooo.bar')
fooobar_logger.addHandler(ch2)
fooobar_logger.propagate = True
fooobar_logger.setLevel(logging.DEBUG)
print('--- fooobar_logger (propagate: False) ---')
fooobar_logger.info('FOOOBAR')
上記コードを実行すると、以下のように出力されます。
--- foo_logger ---
CH1: foo - FOO
--- foobar1_logger (propagate: True) ---
CH2: foo.bar1 - FOOBAR1
CH1: foo.bar1 - FOOBAR1
--- foobar2_logger (propagate: False) ---
CH2: foo.bar2 - FOOBAR2
--- fooobar_logger (propagate: False) ---
CH2: fooo.bar - FOOOBAR
ppyt/commands/dummy.py
__init__.pyのCommandBaseクラスを継承して、dummy.pyにCommandクラスを定義します。
# coding: utf-8
import logging
from ppyt.commands import CommandBase
logger = logging.getLogger(__name__)
plogger = logging.getLogger('print')
class Command(CommandBase):
"""特に意味のないコマンドです。"""
def _add_options(self, parser):
"""コマンド実行時の引数を定義します。"""
parser.add_argument('args', nargs='*')
def _execute(self, options):
logger.debug(self._command)
logger.info('args: {}'.format(options.args))
plogger.info('Hello World!')
_add_optionsメソッドでは、引数のパターンを登録していきます。ここでは、0個以上のハイフン無しの引数を取れるようにしてあります。
_executeメソッドにはコマンド本体の処理を記述していきます。ここではloggerとploggerを使ってメッセージを表示しています。
ppyt/const.pyとppyt/exceptions.py
コマンドを実行するためには、ppyt/const.pyとppyt/exceptions.pyが必要です。
const.pyは定数を定義しています。exceptions.pyは例外クラスを定義したモジュールです。Exceptionを継承していくつかの例外クラスを定義しています。
独自の例外クラスの実装方法
独自の例外クラスは、Exceptionクラスを継承するだけで作成することができます。Exceptionから処理を変更する必要がない場合は何も修正する必要はありません。
class FooError(Exception):
pass
def raise_foo():
raise FooError('RAISE FOO ERROR') # FooErrorを投げます。
もし、メッセージを加工したい場合は、__init__.pyクラスをオーバーライドするだけで対応できます。例えば、受け取ったエラーメッセージの前後に「|」を加える場合は、文字列を加工して親クラス（Exception）のコンストラクタを呼びます。
class BarError(Exception):
def __init__(self, msg):
msg = '|' + msg + '|' # 受け取ったメッセージの前後に「|」をつけます。
super().__init__(msg)
def raise_bar():
raise BarError('RAISE BAR ERROR') # BarErrorを投げます。
下記テストケースを見ると、FooErrorはメッセージがそのまま設定されていて、BarErrorはメッセージの前後に「|」が付いていることがわかります。
import unittest
class TestCase(unittest.TestCase):
def test_foo(self):
self.assertRaises(FooError, raise_foo)
try:
raise_foo()
self.fail('ここには入りません。')
except FooError as err:
self.assertEqual(str(err), 'RAISE FOO ERROR')
except:
self.fail('ここには入りません。')
def test_bar(self):
self.assertRaises(BarError, raise_bar)
try:
raise_bar()
self.fail('ここには入りません。')
except BarError as err:
self.assertEqual(str(err), '|RAISE BAR ERROR|')
except:
self.fail('ここには入りません。')
if __name__ == '__main__':
unittest.main()
コマンドを実行
パッケージのインストール
「requirement」を使ってNumPyとSQLAlchemyをインストールしておきます。
$ pip install -r /path/to/requirement
または、パッケージを指定してインストールします。
$ pip install numpy sqlalchemy
dummyコマンドを実行する。
dummyコマンドは「./manager.py dummy 引数」や「python3 manager.py 引数」のような形式で実行することができます。※環境によって実行方法は異なります。
引数に「a」・「b」・「c」を指定して実行すると、引数を受け取れていることを確認できます。
$ ./manager.py dummy a b c
xxxx-xx-xx xx:xx:xx,xxx - __init__.py.start#52 - INFO - dummyコマンドを開始します。
xxxx-xx-xx xx:xx:xx,xxx xx:xx:xx,xxx - dummy.py._execute#18 - INFO - args: ['a', 'b', 'c']
Hello World!
xxxx-xx-xx xx:xx:xx,xxx xx:xx:xx,xxx - dummy.py._execute#19 - INFO - Hello World!
xxxx-xx-xx xx:xx:xx,xxx - __init__.py.start#55 - INFO - dummyコマンドが正常に終了しました。実行時間: 0.0秒
「-v」か「--verbose」を指定するとDEBUGレベルのログも出力されるようになります。
$ ./manager.py --verbose dummy arg
...
xxxx-xx-xx xx:xx:xx,xxx - dummy.py._execute#17 - DEBUG - dummy
...
また、コマンドを指定しない、または存在しない場合はコマンド一覧が表示されます。
$ ./manager.py list
# コマンド一覧
* dummy
この記事が役に立った場合、シェアしていただけると励みになります！！