Finance

#!/usr/bin/python

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

# http://www.apache.org/licenses/LICENSE-2.0

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

__author__ = 'thesweeheng@gmail.com'

from gdata.finance.service import \

FinanceService, PortfolioQuery, PositionQuery

from gdata.finance import \

PortfolioEntry, PortfolioData, TransactionEntry, TransactionData, \

Price, Commission, Money

import datetime

import sys

def PrintReturns(pfx, d):

"""Print returns."""

print pfx, '%1.5f(1w) %1.5f(4w) %1.5f(3m) %1.5f(YTD)' % tuple(

float(i) for i in (d.return1w, d.return4w, d.return3m, d.returnYTD))

pfx = ' ' * len(pfx)

print pfx, '%1.5f(1y) %1.5f(3y) %1.5f(5y) %1.5f(overall)' % tuple(

float(i) for i in (d.return1y, d.return3y, d.return5y, d.return_overall))

PrRtn = PrintReturns

def PrintTransactions(transactions):

"""Print transactions."""

print " Transactions:"

fmt = ' %4s %-23s %-10s %6s %-11s %-11s'

print fmt % ('ID','Date','Type','Shares','Price','Commission')

for txn in transactions:

d = txn.transaction_data

print fmt % (txn.transaction_id, d.date or '----', d.type,

d.shares, d.price.money[0], d.commission.money[0])

if d.notes:

print " Notes:", d.notes

def PrintPosition(pos, with_returns=False):

"""Print single position."""

print ' Position :', pos.position_title

print ' Ticker ID :', pos.ticker_id

print ' Symbol :', pos.symbol

print ' Last updated :', pos.updated.text

d = pos.position_data

print ' Shares :', d.shares

if with_returns:

print ' Gain % :', d.gain_percentage

PrRtn(' Returns :', d)

print ' Cost basis :', d.cost_basis

print ' Days gain :', d.days_gain

print ' Gain :', d.gain

print ' Market value :', d.market_value

if pos.transactions:

print " <inlined transactions>\n"

PrintTransactions(pos.transactions)

print " </inlined transactions>\n"

def PrintPositions(positions, with_returns=False):

for pos in positions:

PrintPosition(pos, with_returns)

def PrintPortfolio(pfl, with_returns=False):

"""Print single portfolio."""

print 'Portfolio Title :', pfl.portfolio_title

print 'Portfolio ID :', pfl.portfolio_id

print ' Last updated :', pfl.updated.text

d = pfl.portfolio_data

print ' Currency :', d.currency_code

if with_returns:

print ' Gain % :', d.gain_percentage

PrRtn(' Returns :', d)

print ' Cost basis :', d.cost_basis

print ' Days gain :', d.days_gain

print ' Gain :', d.gain

print ' Market value :', d.market_value

if pfl.positions:

print " <inlined positions>\n"

PrintPositions(pfl.positions, with_returns)

print " </inlined positions>\n"

def PrintPortfolios(portfolios, with_returns=False):

for pfl in portfolios:

PrintPortfolio(pfl, with_returns)

def ShowCallDetails(meth):

def wrap(*args, **kwargs):

print '@', meth.__name__, args[1:], kwargs

meth(*args, **kwargs)

return wrap

class FinanceTester(object):

def __init__(self, email, password):

self.client = FinanceService(source='gdata-finance-test')

self.client.ClientLogin(email, password)

def GetPortfolios(self, with_returns=False, inline_positions=False):

query = PortfolioQuery()

query.returns = with_returns

query.positions = inline_positions

return self.client.GetPortfolioFeed(query=query).entry

def GetPositions(self, portfolio, with_returns=False, inline_transactions=False):

query = PositionQuery()

query.returns = with_returns

query.transactions = inline_transactions

return self.client.GetPositionFeed(portfolio, query=query).entry

def GetTransactions(self, position=None, portfolio=None, ticker=None):

if position:

feed = self.client.GetTransactionFeed(position)

elif portfolio and ticker:

feed = self.client.GetTransactionFeed(

portfolio_id=portfolio.portfolio_id, ticker_id=ticker)

return feed.entry

@ShowCallDetails

def TestShowDetails(self, with_returns=False, inline_positions=False,

inline_transactions=False):

portfolios = self.GetPortfolios(with_returns, inline_positions)

for pfl in portfolios:

PrintPortfolio(pfl, with_returns)

positions = self.GetPositions(pfl, with_returns, inline_transactions)

for pos in positions:

PrintPosition(pos, with_returns)

PrintTransactions(self.GetTransactions(pos))

def DeletePortfoliosByName(self, portfolio_titles):

for pfl in self.GetPortfolios():

if pfl.portfolio_title in portfolio_titles:

self.client.DeletePortfolio(pfl)

def AddPortfolio(self, portfolio_title, currency_code):

pfl = PortfolioEntry(portfolio_data=PortfolioData(

currency_code=currency_code))

pfl.portfolio_title = portfolio_title

return self.client.AddPortfolio(pfl)

def UpdatePortfolio(self, portfolio,

portfolio_title=None, currency_code=None):

if portfolio_title:

portfolio.portfolio_title = portfolio_title

if currency_code:

portfolio.portfolio_data.currency_code = currency_code

return self.client.UpdatePortfolio(portfolio)

def DeletePortfolio(self, portfolio):

self.client.DeletePortfolio(portfolio)

@ShowCallDetails

def TestManagePortfolios(self):

pfl_one = 'Portfolio Test: Emerging Markets 12345'

pfl_two = 'Portfolio Test: Renewable Energy 31415'

print '---- Deleting portfolios ----'

self.DeletePortfoliosByName([pfl_one, pfl_two])

PrintPortfolios(self.GetPortfolios())

print '---- Adding new portfolio ----'

pfl = self.AddPortfolio(pfl_one, 'SGD')

PrintPortfolios(self.GetPortfolios())

print '---- Changing portfolio title and currency code ----'

pfl = self.UpdatePortfolio(pfl, pfl_two, 'USD')

PrintPortfolios(self.GetPortfolios())

print '---- Deleting portfolio ----'

self.DeletePortfolio(pfl)

PrintPortfolios(self.GetPortfolios())

def Transact(self, type, portfolio, ticker, date=None, shares=None,

notes=None, price=None, commission=None, currency_code=None):

if price is not None:

price = Price(money=[Money(amount=str(price),

currency_code=currency_code or

portfolio.portfolio_data.currency_code)])

if commission is not None:

commission = Commission(money=[Money(amount=str(comission),

currency_code=currency_code or

portfolio.portfolio_data.currency_code)])

if date is not None and isinstance(date, datetime.datetime):

date = date.isoformat()

if shares is not None:

shares = str(shares)

txn = TransactionEntry(transaction_data=TransactionData(type=type,

date=date, shares=shares, notes=notes, price=price,

commission=commission))

return self.client.AddTransaction(txn,

portfolio_id=portfolio.portfolio_id, ticker_id=ticker)

def Buy(self, portfolio, ticker, **kwargs):

return self.Transact('Buy', portfolio, ticker, **kwargs)

def Sell(self, portfolio, ticker, **kwargs):

return self.Transact('Sell', portfolio, ticker, **kwargs)

def GetPosition(self, portfolio, ticker, with_returns=False, inline_transactions=False):

query = PositionQuery()

query.returns = with_returns

query.transactions = inline_transactions

return self.client.GetPosition(

portfolio_id=portfolio.portfolio_id, ticker_id=ticker, query=query)

def DeletePosition(self, position):

self.client.DeletePosition(position_entry=position)

def UpdateTransaction(self, transaction):

self.client.UpdateTransaction(transaction)

def DeleteTransaction(self, transaction):

self.client.DeleteTransaction(transaction)

@ShowCallDetails

def TestManageTransactions(self):

pfl_title = 'Transaction Test: Technology 27182'

self.DeletePortfoliosByName([pfl_title])

print '---- Adding new portfolio ----'

pfl = self.AddPortfolio(pfl_title, 'USD')

PrintPortfolios(self.GetPortfolios())

print '---- Adding buy transactions ----'

tkr1 = 'NASDAQ:GOOG'

date = datetime.datetime(2009,04,01)

days = datetime.timedelta(1)

txn1 = self.Buy(pfl, tkr1, shares=500, price=321.00, date=date)

txn2 = self.Buy(pfl, tkr1, shares=150, price=312.00, date=date+15*days)

pos = self.GetPosition(portfolio=pfl, ticker=tkr1, with_returns=True)

PrintPosition(pos, with_returns=True)

PrintTransactions(self.GetTransactions(pos))

print '---- Adding sell transactions ----'

txn3 = self.Sell(pfl, tkr1, shares=400, price=322.00, date=date+30*days)

txn4 = self.Sell(pfl, tkr1, shares=200, price=330.00, date=date+45*days)

pos = self.GetPosition(portfolio=pfl, ticker=tkr1, with_returns=True)

PrintPosition(pos, with_returns=True)

PrintTransactions(self.GetTransactions(pos))

print "---- Modifying first and deleting third ----"

txn1.transaction_data.shares = '400.0'

self.UpdateTransaction(txn1)

self.DeleteTransaction(txn3)

pos = self.GetPosition(portfolio=pfl, ticker=tkr1, with_returns=True)

PrintPosition(pos, with_returns=True)

PrintTransactions(self.GetTransactions(pos))

print "---- Deleting position ----"

print "Number of positions (before):", len(self.GetPositions(pfl))

self.DeletePosition(pos)

print "Number of positions (after) :", len(self.GetPositions(pfl))

print '---- Deleting portfolio ----'

self.DeletePortfolio(pfl)

PrintPortfolios(self.GetPortfolios())

if __name__ == '__main__':

try:

email = sys.argv[1]

password = sys.argv[2]

cases = sys.argv[3:]

except IndexError:

print "Usage: test_finance account@google.com password [0 1 2...]"

sys.exit(1)

tester = FinanceTester(email, password)

tests = [

tester.TestShowDetails,

lambda: tester.TestShowDetails(with_returns=True),

tester.TestManagePortfolios,

tester.TestManageTransactions,

lambda: tester.TestShowDetails(with_returns=True, inline_positions=True),

lambda: tester.TestShowDetails(with_returns=True, inline_positions=True,

inline_transactions=True),]

if not cases:

cases = range(len(tests))

for i in cases:

print "===== TEST CASE", i, "="*50

tests[int(i)]()