我的人生目標到底是什麼？我的生存究竟有什麼意義？我開始覺得，不管在什麼地方過活，日子過得在好，錢賺得再多，生意做得再大，我也只是重複眼前的這種生活。
visual_manager.py
import logging
import sys
import src.jaguar_report.visual_indicator as indicator
import src.common.logger as logger
class VisualManager():
def view_template(self, v, args):
assert args.stock_code
if args.log:
v.save(args.stock_code, args.log)
else:
v.show(args.stock_code)
def view_roe(self, args):
self.view_template(indicator.VisualRoe(), args)
def view_financial_structure(self, args):
self.view_template(indicator.VisualFinancialStructure(), args)
def view_current_and_quick_ratio(self, args):
pass
def view_cash_flow(self, args):
self.view_template(indicator.VisualCashFlow(), args)
def main():
logger.config_root(level=logging.DEBUG)
import argparse
parser = argparse.ArgumentParser()
parser.add_argument('-t', '--target', help='set target: roe, \
financial_structure, cash_flow')
parser.add_argument('-s', '--stock_code', help='set stock code')
parser.add_argument('-l', '--log', help='set log file to save visual chart')
args = parser.parse_args()
m = VisualManager()
map = {
'roe': m.view_roe,
'financial_structure': m.view_financial_structure,
'cash_flow': m.view_cash_flow,
}
assert args.target
assert args.target in map
map[args.target](args)
if __name__ == '__main__':
sys.exit(main())
visual_indicator.py
import logging
from ..common import logger
from ..common import plot_util
class VisualBase():
def __init__(self):
self.LOGGER = logging.getLogger()
self.__plot_util = plot_util.PlotUtil()
def get_config(self, stock_code):
pass
def save(self, stock_code, dest_file):
config = self.get_config(stock_code)
try:
self.__plot_util.save_plot(config, dest_file)
except IndexError as e:
self.LOGGER.error(e)
def show(self, stock_code):
config = self.get_config(stock_code)
try:
self.__plot_util.show_plot(config)
except IndexError as e:
self.LOGGER.error(e)
class VisualRoe(VisualBase):
def get_config(self, stock_code):
from ..stocktotal import roe_query
q = roe_query.RoeQuery()
return {
'primary_data_list': [
(q.query_roe(stock_code), 'ROE'),
(q.query_net_profit_margin(stock_code), 'Net Profit Margin'),
(q.query_total_assets_turnover(stock_code), 'Total Assets Turnover'),
],
'secondary_data_list': [
(q.query_equity_multiplier(stock_code), 'Equity Multiplier')
],
'title': '''ROE Analysis: %s''' % stock_code,
'percent_formatter': None,
}
class VisualFinancialStructure(VisualBase):
def get_config(self, stock_code):
from ..stocktotal import financial_structure_query
q = financial_structure_query.FinancialStructureQuery()
return {
'primary_data_list': [
(q.query_equity_ratio(stock_code), 'Equity Ratio'),
(q.query_debt_ratio(stock_code), 'Debt Ratio'),
],
'secondary_data_list': [
(q.query_equity_multiplier(stock_code), 'Equity Multiplier')
],
'title': '''Financial Structure Analysis: %s''' % stock_code,
'percent_formatter': None,
}
class VisualCashFlow(VisualBase):
def get_config(self, stock_code):
from ..stocktotal import cash_flow_query
q = cash_flow_query.CashFlowQuery()
# TODO: operating activity, financing activity, investing activity,
# total cash flow, operating activity - net income
return {
'primary_data_list': [
(q.query_free_cash_flow(stock_code), 'Free Cash Flow'),
],
'title': '''Free Cash Flow Analysis: %s''' % stock_code,
}
plot_util.py
import numpy as np
import matplotlib.pyplot as plt
from matplotlib.ticker import FuncFormatter
def percent_formatter(x, pos=0):
return '''%1.f %%''' % (100*x)
class PlotUtil():
def __init__(self):
self.COLOR_LIST = ['blue', 'red', 'green', 'cyan', 'magenta', 'black', 'yellow']
def show_plot(self, config):
figure, legend = self.plot(config)
plt.show()
plt.close(figure)
def save_plot(self, config, dest_file):
figure, legend = self.plot(config)
figure.savefig(dest_file, bbox_extra_artists=(legend,), bbox_inches='tight')
plt.close(figure)
def plot(self, config):
primary_data_list = config['primary_data_list']
assert primary_data_list
color_iter = iter(self.COLOR_LIST)
figure = plt.figure()
ax1 = figure.add_subplot(111)
if 'percent_formatter' in config:
ax1.yaxis.set_major_formatter(FuncFormatter(percent_formatter))
lines, labels = [], []
for data in primary_data_list:
series = np.core.records.fromrecords(data[0], names='date,number')
x_series = [_.toordinal() for _ in series.date]
y_series = series.number
ax1.plot_date(x=x_series, y=y_series, fmt='-', linewidth=2, \
color=next(color_iter), label=data[1])
lines1, labels1 = ax1.get_legend_handles_labels()
lines, labels = lines + lines1, labels + labels1
if 'secondary_data_list' in config:
secondary_data_list = config['secondary_data_list']
assert secondary_data_list
ax2 = ax1.twinx()
for data in secondary_data_list:
series = np.core.records.fromrecords(data[0], names='date,number')
x_series = [_.toordinal() for _ in series.date]
y_series = series.number
ax2.plot_date(x=x_series, y=y_series, fmt='-', linewidth=2, \
color=next(color_iter), label=data[1])
lines2, labels2 = ax2.get_legend_handles_labels()
lines, labels = lines + lines2, labels + labels2
legend = ax1.legend(lines, labels, loc='upper center', bbox_to_anchor=(0.5, -0.05), ncol=1)
if 'title' in config:
plt.title(config['title'])
plt.grid(True)
return figure, legend
cash_flow_query.py
from . import query
class CashFlowQuery(query.Query):
def __init__(self):
query.Query.__init__(self)
def query_free_cash_flow(self, stock_code):
SQL_SELECT = \
'''
select activity_date, sum(number) as free_cash_flow from
(
select *, max(report_date) from CashFlowStmt where
report_type = 'C'
and stock_code = ?
and item in ('Operating', 'Investing')
group by activity_date, item, stock_code
)
group by activity_date
'''
return self.exec_query_series(SQL_SELECT, stock_code)
roe_query.py
# coding: big5
from . import query
class RoeQuery(query.Query):
def __init__(self):
query.Query.__init__(self)
"""
(annual adjusted) ROE
= net income / shareholder equity
= ( net income / operating income )
x ( operating income / total assets )
x ( total assets / shareholder equity )
= ( net profit margin ) x ( total assets turnover ) x ( equity multiplier )
Refernece: 郭恭克, 獵豹財務長投資魔法書 (ISBN：9789868340091)
"""
def query_roe(self, stock_code):
SQL_SELECT = \
'''
select activity_date,
case
when strftime('%m', activity_date) = '03' then roe * 4/1
when strftime('%m', activity_date) = '06' then roe * 4/2
when strftime('%m', activity_date) = '09' then roe * 4/3
else roe
end as annual_adjusted_roe
from
(
select activity_date, roe, max(report_date) from
(
select
E.activity_date,
I.number / E.number as roe,
E.report_date
from BalanceSheet as E
inner join
IncomeStmt as I
on E.stock_code = I.stock_code
and E.activity_date = I.activity_date
and E.item = '股東權益總計'
and I.item = '合併總損益'
and E.report_type = 'C'
and I.report_type = 'C'
and E.stock_code = ?
)
where roe is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
"""
net profit margin = net income / operating income
"""
def query_net_profit_margin(self, stock_code):
SQL_SELECT = \
'''
select activity_date, net_profit_margin from
(
select activity_date, net_profit_margin, max(report_date) from
(
select
I.activity_date,
I.number / O.number as net_profit_margin,
I.report_date
from IncomeStmt as I
inner join
IncomeStmt as O
on I.stock_code = O.stock_code
and I.activity_date = O.activity_date
and I.item = '合併總損益'
and O.item = '營業收入合計'
and I.report_type = 'C'
and O.report_type = 'C'
and I.stock_code = ?
)
where net_profit_margin is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
"""
(annual adjusted) total assets turnover = operating income / total assets
"""
def query_total_assets_turnover(self, stock_code):
SQL_SELECT = \
'''
select activity_date,
case
when strftime('%m', activity_date) = '03' then total_assets_turnover * 4/1
when strftime('%m', activity_date) = '06' then total_assets_turnover * 4/2
when strftime('%m', activity_date) = '09' then total_assets_turnover * 4/3
else total_assets_turnover
end as annual_adjusted_total_assets_turnover
from
(
select activity_date, total_assets_turnover, max(report_date) from
(
select
A.activity_date,
O.number / A.number as total_assets_turnover,
A.report_date
from BalanceSheet as A
inner join
IncomeStmt as O
on A.stock_code = O.stock_code
and A.activity_date = O.activity_date
and A.item = '資產總計'
and O.item = '營業收入合計'
and A.report_type = 'C'
and O.report_type = 'C'
and A.stock_code = ?
)
where total_assets_turnover is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
"""
equity multiplier = total assets / shareholder equity
"""
def query_equity_multiplier(self, stock_code):
SQL_SELECT = \
'''
select activity_date, equity_multiplier from
(
select activity_date, equity_multiplier, max(report_date) from
(
select
E.activity_date,
A.number / E.number as equity_multiplier,
E.report_date
from BalanceSheet as E
inner join
BalanceSheet as A
on E.stock_code = A.stock_code
and E.activity_date = A.activity_date
and E.item = '股東權益總計'
and A.item = '資產總計'
and E.report_type = 'C'
and A.report_type = 'C'
and E.stock_code = ?
)
where equity_multiplier is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
financial_structure_query.py
# coding: big5
from . import query
class FinancialStructureQuery(query.Query):
def __init__(self):
query.Query.__init__(self)
def query_equity_ratio(self, stock_code):
SQL_SELECT = \
'''
select activity_date, equity_ratio from
(
select activity_date, equity_ratio, max(report_date) from
(
select
E.activity_date,
E.number / A.number as equity_ratio,
E.report_date
from BalanceSheet as E
inner join
BalanceSheet as A
on E.stock_code = A.stock_code
and E.activity_date = A.activity_date
and E.item = '股東權益總計'
and A.item = '資產總計'
and E.report_type = 'C'
and A.report_type = 'C'
and E.stock_code = ?
)
where equity_ratio is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
def query_debt_ratio(self, stock_code):
SQL_SELECT = \
'''
select activity_date, debt_ratio from
(
select activity_date, debt_ratio, max(report_date) from
(
select
D.activity_date,
D.number / A.number as debt_ratio,
D.report_date
from BalanceSheet as D
inner join
BalanceSheet as A
on D.stock_code = A.stock_code
and D.activity_date = A.activity_date
and D.item = '負債總計'
and A.item = '資產總計'
and D.report_type = 'C'
and A.report_type = 'C'
and D.stock_code = ?
)
where debt_ratio is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
def query_equity_multiplier(self, stock_code):
SQL_SELECT = \
'''
select activity_date, equity_multiplier from
(
select activity_date, equity_multiplier, max(report_date) from
(
select
E.activity_date,
A.number / E.number as equity_multiplier,
E.report_date
from BalanceSheet as E
inner join
BalanceSheet as A
on E.stock_code = A.stock_code
and E.activity_date = A.activity_date
and E.item = '股東權益總計'
and A.item = '資產總計'
and E.report_type = 'C'
and A.report_type = 'C'
and E.stock_code = ?
)
where equity_multiplier is not null
group by activity_date
order by activity_date
)
'''
return self.exec_query_series(SQL_SELECT, stock_code)
query.py
import os
import sqlite3
from datetime import datetime
class Query():
def __init__(self):
self.DB_FILE = './db/stocktotal.db'
assert os.path.isfile(self.DB_FILE)
def exec_query_series(self, sql_cmd, stock_code):
conn = sqlite3.connect(self.DB_FILE)
cursor = conn.cursor()
cursor.execute(sql_cmd, [stock_code])
rv = [(datetime.strptime(_[0], '%Y-%m-%d'), _[1]) for _ in cursor.fetchall()]
cursor.close()
conn.close()
return rv
def exec_query_stock_code(self, sql_cmd):
conn = sqlite3.connect(self.DB_FILE)
cursor = conn.cursor()
cursor.execute(sql_cmd)
rv = [_[0] for _ in cursor.fetchall()]
cursor.close()
conn.close()
return rv
沒有留言:
張貼留言