from __future__ import division
from __future__ import unicode_literals
import pytz
_type = type
import datetime
from decimal import Decimal
from sqlalchemy import Column, VARCHAR, INTEGER, ForeignKey, BIGINT, Index
from sqlalchemy.orm import relation
from sqlalchemy.orm.exc import MultipleResultsFound
from ._commodity_helper import quandl_fx
from .._common import CallableList, GncConversionError
from .._common import GnucashException, hybrid_property_gncnumeric
from .._declbase import DeclarativeBaseGuid
from ..sa_extra import _DateAsDateTime
from ..yahoo_client import get_latest_quote, download_quote
[docs]class GncCommodityError(GnucashException):
pass
[docs]class GncPriceError(GnucashException):
pass
[docs]class Price(DeclarativeBaseGuid):
"""
A single Price for a commodity.
Attributes:
commodity (:class:`Commodity`): commodity to which the Price relates
currency (:class:`Commodity`): currency in which the Price is expressed
date (:class:`datetime.date`): date object representing the day at which the price is relevant
source (str): source of the price
type (str): last, ask, bid, unknown, nav
value (:class:`decimal.Decimal`): the price itself
"""
__tablename__ = "prices"
__table_args__ = {}
# column definitions
commodity_guid = Column(
"commodity_guid",
VARCHAR(length=32),
ForeignKey("commodities.guid"),
nullable=False,
)
currency_guid = Column(
"currency_guid",
VARCHAR(length=32),
ForeignKey("commodities.guid"),
nullable=False,
)
date = Column("date", _DateAsDateTime(neutral_time=False), nullable=False)
source = Column("source", VARCHAR(length=2048))
type = Column("type", VARCHAR(length=2048))
_value_num = Column("value_num", BIGINT(), nullable=False)
_value_denom = Column("value_denom", BIGINT(), nullable=False)
value = hybrid_property_gncnumeric(_value_num, _value_denom)
# relation definitions
commodity = relation(
"Commodity",
back_populates="prices",
foreign_keys=[commodity_guid],
)
currency = relation(
"Commodity",
foreign_keys=[currency_guid],
)
def __init__(
self, commodity, currency, date, value, type="unknown", source="user:price"
):
self.commodity = commodity
self.currency = currency
assert _type(date) is datetime.date
self.date = date
self.value = value
self.type = type
self.source = source
def __str__(self):
return "Price<{:%Y-%m-%d} : {} {}/{}>".format(
self.date, self.value, self.currency.mnemonic, self.commodity.mnemonic
)
[docs] def object_to_validate(self, change):
if change[-1] != "deleted":
yield self
[docs] def validate(self):
# check uniqueness of namespace/mnemonic
try:
self.book.query(Price).filter_by(
commodity=self.commodity,
currency=self.currency,
source=self.source,
date=self.date,
).one()
except MultipleResultsFound:
raise ValueError("{} already exists in this book".format(self))
[docs]class Commodity(DeclarativeBaseGuid):
"""
A GnuCash Commodity.
Attributes:
cusip (str): cusip code
fraction (int): minimal unit of the commodity (e.g. 100 for 1/100)
namespace (str): CURRENCY for currencies, otherwise any string to group multiple commodities together
mnemonic (str): the ISO symbol for a currency or the stock symbol for stocks (used for online quotes)
quote_flag (int): 1 if piecash/GnuCash quotes will retrieve online quotes for the commodity
quote_source (str): the quote source for GnuCash (piecash always use yahoo for stock and quandl for currencies
quote_tz (str): the timezone to assign on the online quotes
base_currency (:class:`Commodity`): The base_currency for a commodity:
- if the commodity is a currency, returns the "default currency" of the book (ie the one of the root_account)
- if the commodity is not a currency, returns the currency encoded in the quoted_currency slot
accounts (list of :class:`piecash.core.account.Account`): list of accounts which have the commodity as commodity
transactions (list of :class:`piecash.core.transaction.Transaction`): list of transactions which have the commodity as currency
prices (iterator of :class:`Price`): iterator on prices related to the commodity (it is a sqlalchemy query underneath)
"""
__tablename__ = "commodities"
__table_args__ = (
Index(
"_unique_cdty",
"namespace",
"mnemonic",
unique=True,
mysql_length={"namespace": 200, "mnemonic": 10},
),
)
# column definitions
namespace = Column("namespace", VARCHAR(length=2048), nullable=False)
mnemonic = Column("mnemonic", VARCHAR(length=2048), nullable=False)
fullname = Column("fullname", VARCHAR(length=2048))
cusip = Column("cusip", VARCHAR(length=2048))
fraction = Column("fraction", INTEGER(), nullable=False)
quote_flag = Column("quote_flag", INTEGER(), nullable=False)
quote_source = Column("quote_source", VARCHAR(length=2048))
quote_tz = Column("quote_tz", VARCHAR(length=2048))
@property
def base_currency(self):
b = self.book
if b is None:
raise GnucashException(
"The commodity should be linked to a session to have a 'base_currency'"
)
if self.namespace == "CURRENCY":
# get the base currency as first commodity in DB
return b.default_currency
else:
# retrieve currency from quoted_currency kvp
# TODO: recover from the web (as fallback)
mnemonic = self.get("quoted_currency", None)
if mnemonic:
return b.currencies(mnemonic=mnemonic)
else:
raise GnucashException(
"The commodity '{}' has no information about its base currency. "
"Add a kvp item named 'quoted_currency' with the mnemonic of the "
"currency to have proper behavior".format(self.mnemonic)
)
# relation definitions
accounts = relation(
"Account",
back_populates="commodity",
cascade="all, delete-orphan",
collection_class=CallableList,
)
transactions = relation(
"Transaction",
back_populates="currency",
cascade="all, delete-orphan",
collection_class=CallableList,
)
prices = relation(
"Price",
back_populates="commodity",
foreign_keys=[Price.commodity_guid],
cascade="all, delete-orphan",
lazy="dynamic",
)
def __init__(
self,
namespace,
mnemonic,
fullname,
fraction=100,
cusip="",
quote_flag=0,
quote_source=None,
quote_tz="",
book=None,
):
if quote_source is None:
quote_source = "currency" if namespace == "CURRENCY" else "yahoo"
if book is not None:
book.add(self)
self.namespace = namespace
self.mnemonic = mnemonic
self.fullname = fullname
self.fraction = fraction
self.cusip = cusip
self.quote_flag = quote_flag
self.quote_source = quote_source
self.quote_tz = quote_tz
if book is not None:
book.flush()
def __str__(self):
return "Commodity<{}:{}>".format(self.namespace, self.mnemonic)
@property
def precision(self):
return len(str(self.fraction)) - 1
[docs] def currency_conversion(self, currency):
"""
Return the latest conversion factor to convert self to currency
Attributes:
currency (:class:`piecash.core.commodity.Commodity`): the currency to which the Price need to be converted
Returns:
a Decimal that can be multiplied by an amount expressed in self.commodity to get an amount expressed in currency
Raises:
GncConversionError: not possible to convert self to the currency
"""
# conversion is done from self.commodity to commodity (if possible)
sc2c = (
self.prices.filter_by(currency=currency).order_by(Price.date.desc()).first()
)
if sc2c:
return sc2c.value
# conversion is done directly from commodity to self.commodity (if possible)
c2sc = (
currency.prices.filter_by(currency=self).order_by(Price.date.desc()).first()
)
if c2sc:
return Decimal(1) / c2sc.value
raise GncConversionError("Cannot convert {} to {}".format(self, currency))
[docs] def update_prices(self, start_date=None):
"""
Retrieve online prices for the commodity:
- for currencies, it will get from quandl the exchange rates between the currency and its base_currency
- for stocks, it will get from yahoo the daily closing prices expressed in its base_currency
Args:
start_date (:class:`datetime.date`): prices will be updated as of the start_date. If None, start_date is today
- 7 days.
.. note:: if prices are already available in the GnuCash file, the function will only retrieve prices as of the
max(start_date, last quoted price date)
.. todo:: add some frequency to retrieve prices only every X (week, month, ...)
"""
if self.book is None:
raise GncPriceError(
"Cannot update price for a commodity not attached to a book"
)
# get last_price updated
last_price = self.prices.order_by(Price.date.desc()).limit(1).first()
if start_date is None:
start_date = datetime.datetime.today().date() + datetime.timedelta(days=-7)
if last_price:
start_date = max(last_price.date + datetime.timedelta(days=1), start_date)
if self.namespace == "CURRENCY":
# get reference currency (from book.root_account)
default_currency = self.base_currency
if default_currency == self:
raise GncPriceError("Cannot update exchange rate for base currency")
# through Quandl for exchange rates
quotes = quandl_fx(self.mnemonic, default_currency.mnemonic, start_date)
for q in quotes:
p = Price(
commodity=self,
currency=default_currency,
date=datetime.datetime.strptime(q.date, "%Y-%m-%d").date(),
value=str(q.rate),
)
else:
symbol = self.mnemonic
share = get_latest_quote(symbol)
currency = self.book.currencies(mnemonic=share.currency)
tz = pytz.timezone(share.timezone)
# get historical data
for q in download_quote(symbol, start_date, datetime.date.today(), tz):
Price(
commodity=self,
currency=currency,
date=q.date,
value=q.close,
type="last",
)
[docs] def object_to_validate(self, change):
if change[-1] != "deleted":
yield self
[docs] def validate(self):
# check uniqueness of namespace/mnemonic
try:
self.book.query(Commodity).filter_by(
namespace=self.namespace, mnemonic=self.mnemonic
).one()
except MultipleResultsFound:
raise ValueError("{} already exists in this book".format(self))