pax_global_header 0000666 0000000 0000000 00000000064 14575653726 0014536 g ustar 00root root 0000000 0000000 52 comment=5f3d558793b537a74480241ac6981479f5938cd3
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/ 0000775 0000000 0000000 00000000000 14575653726 0023313 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/ 0000775 0000000 0000000 00000000000 14575653726 0024763 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/ 0000775 0000000 0000000 00000000000 14575653726 0026234 5 ustar 00root root 0000000 0000000 woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/__init__.py 0000664 0000000 0000000 00000001505 14575653726 0030346 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012-2020 Budget Insight
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see .
from .module import DegiroModule
__all__ = ['DegiroModule']
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/browser.py 0000664 0000000 0000000 00000040273 14575653726 0030277 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012-2020 Budget Insight
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see .
import datetime
from decimal import Decimal
from dateutil.relativedelta import relativedelta
from requests.exceptions import ConnectionError
from woob.browser import URL, need_login
from woob.browser.mfa import TwoFactorBrowser
from woob.browser.exceptions import (
BrowserTooManyRequests, ClientError, ServerError,
)
from woob.exceptions import (
ActionNeeded, ActionType, BrowserIncorrectPassword, BrowserPasswordExpired,
OfflineOTPQuestion, BrowserUnavailable,
)
from woob.tools.capabilities.bank.investments import create_french_liquidity
from woob.capabilities.base import Currency, empty
from woob.capabilities.bank import Account
from woob.tools.decorators import retry
from .pages import (
LoginPage, OtpPage, AccountsPage, AccountDetailsPage,
InvestmentPage, HistoryPage, MarketOrdersPage,
ExchangesPage, MaintenancePage,
)
class URLWithDate(URL):
def go(self, fromDate, toDate=None, *args, **kwargs):
toDate_ = toDate or datetime.datetime.now().strftime('%d/%m/%Y')
return super(URLWithDate, self).go(
toDate=toDate_,
fromDate=fromDate,
account_id=self.browser.int_account,
session_id=self.browser.session_id,
headers={'Accept': 'application/json, text/plain, */*'},
)
class DegiroBrowser(TwoFactorBrowser):
BASEURL = 'https://trader.degiro.nl'
TIMEOUT = 60 # Market orders queries can take a long time
HAS_CREDENTIALS_ONLY = True
maintenance = URL(r'https://www.degiro.nl/maintenance/', MaintenancePage)
login = URL(r'/login/secure/login', LoginPage)
send_otp = URL(r'/login/secure/login/totp', OtpPage)
client = URL(r'/pa/secure/client\?sessionId=(?P.*)', LoginPage)
product = URL(r'/product_search/secure/v5/products/info\?sessionId=(?P.*)', InvestmentPage)
exchanges = URL(r'/product_search/config/dictionary/', ExchangesPage)
accounts = URL(
r'/trading(?P\w*)/secure/v5/update/(?P.*);jsessionid=(?P.*)\?historicalOrders=0' +
r'&orders=0&portfolio=0&totalPortfolio=0&transactions=0&alerts=0&cashFunds=0¤cyExchange=0&',
AccountsPage
)
account_details = URL(
r'https://trader.degiro.nl/trading(?P\w*)/secure/v5/account/info/(?P.*);jsessionid=(?P.*)',
AccountDetailsPage
)
transaction_investments = URLWithDate(
r'/reporting/secure/v4/transactions\?fromDate=(?P.*)' +
'&groupTransactionsByOrder=false&intAccount=(?P.*)' +
'&orderId=&product=&sessionId=(?P.*)' +
'&toDate=(?P.*)',
HistoryPage
)
history = URLWithDate(
r'/reporting/secure/v6/accountoverview\?fromDate=(?P.*)' +
'&groupTransactionsByOrder=false&intAccount=(?P.*)' +
'&orderId=&product=&sessionId=(?P.*)&toDate=(?P.*)',
HistoryPage
)
market_orders = URLWithDate(
r'/reporting/secure/v4/order-history\?fromDate=(?P.*)' +
'&toDate=(?P.*)&intAccount=(?P.*)&sessionId=(?P.*)',
MarketOrdersPage
)
__states__ = ('staging', 'session_id', 'int_account', 'name')
def __init__(self, config, *args, **kwargs):
super(DegiroBrowser, self).__init__(config, *args, **kwargs)
self.staging = None
self.int_account = None
self.name = None
self.session_id = None
self.account = None
self.invs = {}
self.trs = {}
self.products = {}
self.stock_market_exchanges = {}
self.AUTHENTICATION_METHODS = {
'otp': self.handle_otp,
}
def locate_browser(self, state):
# We must check 'staging' & 'session_id' states are set before trying to go on AccountsPage.
if not self.session_id:
return
if not state.get('staging'):
self.staging = ''
if 'staging' in self.session_id:
self.staging = '_s'
try:
# We try reloading the session with the previous states if they are not expired.
# If they are, we encounter a ClientError 401 Unauthorized, we need to relogin.
self.accounts.go(staging=self.staging, account_id=self.int_account, session_id=self.session_id)
except ClientError as e:
if e.response.status_code != 401:
raise
@retry(BrowserTooManyRequests, delay=30)
def init_login(self):
try:
self.login.go(json={'username': self.username, 'password': self.password})
except ClientError as e:
if e.response.status_code == 400:
raise BrowserIncorrectPassword()
elif e.response.status_code == 403:
status = e.response.json().get('statusText', '')
if status == 'accountBlocked':
raise BrowserIncorrectPassword('Your credentials are invalid and your account is currently blocked.')
raise Exception('Login failed with status: "%s".', status)
elif e.response.status_code == 412:
status = e.response.json().get('statusText')
# https://trader.degiro.nl/translations/?language=fr&country=FR&modules=commonFE%2CloginFE
# for status_messages
if status == 'jointAccountPersonNeeded':
# After the first post in a joint account, we get a json containing IDs of
# the account holders. Then we need to make a second post to send the
# ID of the user trying to log in.
persons = e.response.json().get('persons')
if not persons:
raise AssertionError('No profiles to select from')
self.login.go(json={
'password': self.password,
'personId': persons[0]['id'],
'username': self.username,
})
elif status == 'passwordReset':
raise BrowserPasswordExpired("Un e-mail vous a été envoyé afin de réinitialiser votre mot de passe. Veuillez consulter votre boite de réception. Si vous n’êtes pas à l’origine de cette demande, merci de contacter notre service clients.")
elif status:
raise AssertionError('Unhandled status: %s' % status)
else:
raise
elif e.response.status_code == 429:
# We sometimes get an HTTP 429 status code when logging in,
# with no other information than 'Too Many Requests'.
# We want to try again in this case.
raise BrowserTooManyRequests()
else:
raise
if self.maintenance.is_here():
raise BrowserUnavailable()
# if 2FA is required for this user
if self.page.has_2fa():
self.check_interactive()
# An authenticator is used here, so no notification or SMS,
# we use the same message as on the website.
raise OfflineOTPQuestion(
'otp',
message='Enter your confirmation code',
)
else:
self.finalize_login()
def handle_otp(self):
data = {
'oneTimePassword': self.config['otp'].get(),
'password': self.password,
'queryParams': {
'redirectUrl': 'https://trader.degiro.nl/trader/#/markets?enabledLanguageCodes=fr&hasPortfolio=false&favouritesListsIds='
},
'username': self.username.lower(),
}
try:
self.send_otp.go(json=data)
except ClientError as e:
json_err = e.response.json().get('statusText')
if e.response.status_code == 400 and json_err == 'badCredentials':
raise BrowserIncorrectPassword('The confirmation code is incorrect', bad_fields=['otp'])
raise
self.finalize_login()
def finalize_login(self):
self.session_id = self.page.get_session_id()
if not self.session_id:
raise AssertionError(
'Missing a session identifier when finalizing the login.',
)
self.staging = ''
if 'staging' in self.session_id:
self.staging = '_s'
self.client.go(session_id=self.session_id)
self.int_account = self.page.get_information('intAccount')
self.name = self.page.get_information('displayName')
if self.int_account is None:
# For various ActionNeeded, field intAccount is not present in the json.
raise ActionNeeded(
locale="fr-FR", message="Merci de compléter votre profil sur le site de Degiro",
action_type=ActionType.FILL_KYC,
)
def fill_stock_market_exchanges(self):
if not self.stock_market_exchanges:
self.exchanges.go()
self.stock_market_exchanges = self.page.get_stock_market_exchanges()
@need_login
def iter_accounts(self):
if self.account is None:
self.accounts.go(staging=self.staging, account_id=self.int_account, session_id=self.session_id)
self.account = self.page.get_account()
# Go to account details to fetch the right currency
try:
self.account_details.go(staging=self.staging, account_id=self.int_account, session_id=self.session_id)
except ClientError as e:
if e.response.status_code == 412:
# No useful message on the API response. On the website, there is a form to complete after login.
raise ActionNeeded(
locale="fr-FR", message="Merci de compléter votre profil sur le site de Degiro",
action_type=ActionType.FILL_KYC,
)
raise
self.account.currency = self.page.get_currency()
# Account balance is the sum of investments valuations
self.account.balance = sum(inv.valuation.quantize(Decimal('0.00')) for inv in self.iter_investment(self.account))
yield self.account
@need_login
def iter_investment(self, account):
self.fill_stock_market_exchanges()
if account.id not in self.invs:
self.accounts.stay_or_go(staging=self.staging, account_id=self.int_account, session_id=self.session_id)
raw_invests = list(self.page.iter_investment(currency=account.currency, exchanges=self.stock_market_exchanges))
# Some invests are present twice. We need to combine them into one, as it's done on the website.
invests = {}
for raw_inv in raw_invests:
if raw_inv.label not in invests:
invests[raw_inv.label] = raw_inv
else:
invests[raw_inv.label].quantity += raw_inv.quantity
invests[raw_inv.label].valuation += raw_inv.valuation
for inv in invests.values():
# Replace as liquidities investments that are cash
if len(inv.label) < 4 and Currency.get_currency(inv.label):
yield create_french_liquidity(inv.valuation)
# Since we are adding Buy/sell positions of the investments
# We need to filter out investments with a quantity sum equal to 0
# those investments are considered as "closed" on the website
elif empty(inv.quantity) or inv.quantity:
yield inv
@need_login
def fetch_market_orders(self, from_date, to_date):
self.fill_stock_market_exchanges()
market_orders = []
self.market_orders.go(fromDate=from_date.strftime('%d/%m/%Y'), toDate=to_date.strftime('%d/%m/%Y'))
# Market orders are displayed chronogically so we must reverse them
for market_order in sorted(
self.page.iter_market_orders(exchanges=self.stock_market_exchanges),
reverse=True,
key=lambda order: order.date
):
market_orders.append(market_order)
return market_orders
@need_login
def iter_market_orders(self, account):
if account.type not in (Account.TYPE_MARKET, Account.TYPE_PEA):
return
# We can fetch up to 3 months of history (minus one day), older than that the JSON response is empty
# We must fetch market orders 2 weeks at a time because if we fetch too many orders at a time the API crashes
market_orders = []
to_date = datetime.datetime.now()
oldest = (to_date - relativedelta(months=3) + relativedelta(days=1))
step = relativedelta(weeks=2)
from_date = (to_date - step)
while to_date > oldest:
try:
market_orders.extend(self.fetch_market_orders(from_date, to_date))
except (ConnectionError, ServerError):
# Fetching market orders can be impossible within the timeout limit because of there are too many.
# Since we can't fetch 3 months of available market order history with a 2-weeks step,
# we will fetch 2 weeks market order history (by editing 'oldest') and set the 'step' to 2 days.
# That way, we will still fetch recent orders within a reasonable amount of time and prevent any crash.
oldest = (to_date - relativedelta(weeks=2) + relativedelta(days=1))
step = relativedelta(days=2)
from_date = (to_date - step)
market_orders.extend(self.fetch_market_orders(from_date, to_date))
to_date = from_date - relativedelta(days=1)
from_date = max(
oldest,
to_date - step,
)
return market_orders
@need_login
def iter_history(self, account):
if account.id not in self.trs:
fromDate = (datetime.datetime.now() - relativedelta(years=1)).strftime('%d/%m/%Y')
self.transaction_investments.go(fromDate=fromDate)
self.fetch_products(list(self.page.get_products()))
transaction_investments = list(self.page.iter_transaction_investments())
self.history.go(fromDate=fromDate)
# the list can be pretty big, and the tr list too
# avoid doing O(n*n) operation
trinv_dict = {(inv.code, inv._action, inv._datetime): inv for inv in transaction_investments}
trs = list(self.page.iter_history(transaction_investments=NoCopy(trinv_dict), account_currency=account.currency))
self.trs[account.id] = trs
return self.trs[account.id]
# We can encounter random 502 (Bad Gateway), retrying fixes the issue.
@retry(ServerError)
def fetch_products(self, ids):
ids = list(set(ids) - set(self.products.keys()))
if ids:
page = self.product.open(
json=ids,
session_id=self.session_id,
)
self.products.update(page.get_products())
# We can encounter random 502 (Bad Gateway), retrying fixes the issue.
@retry(ServerError)
def get_product(self, id):
if id not in self.products:
self.fetch_products([id])
return self.products[id]
class NoCopy(object):
# params passed to a @method are deepcopied, in each iteration of ItemElement
# so we want to avoid repeatedly copying objects since we don't intend to modify them
def __init__(self, v):
self.v = v
def __deepcopy__(self, memo):
return self
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/module.py 0000664 0000000 0000000 00000004037 14575653726 0030077 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012-2020 Budget Insight
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see .
from woob.tools.backend import Module, BackendConfig
from woob.tools.value import ValueBackendPassword, ValueTransient
from woob.capabilities.bank.wealth import CapBankWealth
from .browser import DegiroBrowser
__all__ = ['DegiroModule']
class DegiroModule(Module, CapBankWealth):
NAME = 'degiro'
DESCRIPTION = u'De giro'
MAINTAINER = u'Jean Walrave'
EMAIL = 'jwalrave@budget-insight.com'
LICENSE = 'LGPLv3+'
VERSION = '3.6'
CONFIG = BackendConfig(
ValueBackendPassword('login', label='Nom d\'utilisateur', masked=False),
ValueBackendPassword('password', label='Mot de passe'),
ValueTransient('otp', regexp=r'\d{6}'),
ValueTransient('request_information'),
)
BROWSER = DegiroBrowser
def create_default_browser(self):
return self.create_browser(
self.config,
self.config['login'].get(),
self.config['password'].get()
)
def iter_accounts(self):
return self.browser.iter_accounts()
def iter_history(self, account):
return self.browser.iter_history(account)
def iter_investment(self, account):
return self.browser.iter_investment(account)
def iter_market_orders(self, account):
return self.browser.iter_market_orders(account)
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/pages.py 0000664 0000000 0000000 00000036657 14575653726 0027726 0 ustar 00root root 0000000 0000000 # -*- coding: utf-8 -*-
# Copyright(C) 2012-2020 Budget Insight
#
# This file is part of a woob module.
#
# This woob module is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This woob module is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this woob module. If not, see .
from decimal import Decimal
import re
from woob.browser.pages import HTMLPage, JsonPage, LoggedPage, RawPage
from woob.browser.elements import ItemElement, DictElement, method
from woob.browser.filters.standard import (
CleanText, Date, Regexp, CleanDecimal,
Env, Field, Currency, Map, Title,
)
from woob.browser.filters.json import Dict
from woob.capabilities.bank import Account
from woob.capabilities.bank.wealth import (
Investment, MarketOrder, MarketOrderDirection, MarketOrderType,
)
from woob.capabilities.base import NotAvailable, empty
from woob.tools.capabilities.bank.transactions import FrenchTransaction
from woob.tools.capabilities.bank.investments import is_isin_valid, IsinCode
def float_to_decimal(f):
return Decimal(str(f))
class MaintenancePage(HTMLPage):
pass
class LoginPage(JsonPage):
def has_2fa(self):
return Dict('statusText', default="")(self.doc) == "totpNeeded"
def get_session_id(self):
return Dict('sessionId')(self.doc)
def get_information(self, information):
key = 'data/' + information
return Dict(key, default=None)(self.doc)
class OtpPage(RawPage):
pass
def list_to_dict(l):
return {d['name']: d.get('value') for d in l}
# Specific currencies are displayed with a factor
# in the API so we must divide the invest valuations
SPECIFIC_CURRENCIES = {
'JPY': 100,
}
class AccountsPage(LoggedPage, JsonPage):
@method
class get_account(ItemElement):
klass = Account
# account balance will be filled with the
# sum of its investments in browser.py
def obj_id(self):
return str(self.page.browser.int_account)
def obj_number(self):
return str(self.page.browser.int_account)
def obj_label(self):
return '%s DEGIRO' % self.page.browser.name.title()
def obj_type(self):
return Account.TYPE_MARKET
@method
class iter_investment(DictElement):
item_xpath = 'portfolio/value'
class item(ItemElement):
klass = Investment
def condition(self):
return float_to_decimal(list_to_dict(self.el['value'])['size'])
obj_unitvalue = Env('unitvalue', default=NotAvailable)
obj_unitprice = Env('unitprice', default=NotAvailable)
obj_original_currency = Env('original_currency', default=NotAvailable)
obj_original_unitvalue = Env('original_unitvalue', default=NotAvailable)
obj_original_unitprice = Env('original_unitprice', default=NotAvailable)
obj_valuation = Env('valuation')
obj_quantity = Env('quantity', default=NotAvailable)
obj_diff = Env('diff')
obj_diff_ratio = Env('diff_ratio', default=NotAvailable)
def obj__product_id(self):
return str(list_to_dict(self.el['value'])['id'])
def obj_label(self):
product_data = Field('_product_data')(self)
return product_data['name']
def obj_vdate(self):
product_data = Field('_product_data')(self)
vdate = product_data.get('closePriceDate') # .get() because certain invest don't have that key in the json
if vdate:
return Date().filter(vdate)
return NotAvailable
def obj_code(self):
product_data = Field('_product_data')(self)
code = product_data['isin']
if is_isin_valid(code):
# Prefix CFD (Contrats for difference) ISIN codes with "XX-"
# to avoid id_security duplicates in the database
if "- CFD" in Field('label')(self):
return "XX-" + code
return code
return NotAvailable
def obj_code_type(self):
if empty(Field('code')(self)):
return NotAvailable
return Investment.CODE_TYPE_ISIN
def obj__product_data(self):
return self.page.browser.get_product(str(Field('_product_id')(self)))
def obj_stock_symbol(self):
product_data = Field('_product_data')(self)
return CleanText(default=NotAvailable).filter(product_data.get('symbol'))
def obj_stock_market(self):
exchanges = Env('exchanges')(self)
product_data = Field('_product_data')(self)
exchange_id = product_data['exchangeId']
if exchange_id:
return exchanges.get(int(exchange_id), NotAvailable)
return NotAvailable
def parse(self, el):
product_data = Field('_product_data')(self)
currency = product_data['currency']
unitvalue = Decimal.quantize(Decimal(list_to_dict(Dict('value')(el))['price']), Decimal('0.0001'))
unitprice = Decimal.quantize(Decimal(list_to_dict(Dict('value')(el))['breakEvenPrice']), Decimal('0.0001'))
quantity = Decimal.quantize(Decimal(list_to_dict(Dict('value')(el))['size']), Decimal('0.01'))
valuation = Decimal(list_to_dict(Dict('value')(el))['value'])
invested_amount = Decimal(list_to_dict(Dict('value')(el))['plBase'][self.env['currency']])
current_valuation = Decimal(list_to_dict(Dict('value')(el))['todayPlBase'][self.env['currency']])
self.env['diff'] = Decimal.quantize(invested_amount - current_valuation, Decimal('0.0001'),)
if invested_amount:
# invested amount can be 0
self.env['diff_ratio'] = Decimal.quantize(self.env['diff'] / abs(invested_amount), Decimal('0.0001'))
if currency == 'GBX':
# Some stocks are priced in GBX (penny sterling)
# We convert them to GBP to avoid ambiguity with the crypto-currency with the same symbol
currency = 'GBP'
unitvalue = unitvalue / 100
unitprice = unitprice / 100
self.env['valuation'] = round(valuation / SPECIFIC_CURRENCIES.get(currency, 1), 2)
self.env['quantity'] = quantity
if currency == self.env['currency']:
self.env['unitvalue'] = unitvalue
self.env['unitprice'] = unitprice
else:
self.env['original_unitvalue'] = unitvalue
self.env['original_unitprice'] = unitprice
self.env['original_currency'] = currency
class AccountDetailsPage(LoggedPage, JsonPage):
def get_currency(self):
return Currency(Dict('data/baseCurrency'))(self.doc)
class InvestmentPage(LoggedPage, JsonPage):
def get_products(self):
return self.doc.get('data', [])
MARKET_ORDER_TYPES = {
0: MarketOrderType.LIMIT,
1: MarketOrderType.MARKET,
2: MarketOrderType.MARKET,
3: MarketOrderType.MARKET,
}
MARKET_ORDER_DIRECTIONS = {
'B': MarketOrderDirection.BUY,
'S': MarketOrderDirection.SALE,
}
class MarketOrdersPage(LoggedPage, JsonPage):
@method
class iter_market_orders(DictElement):
item_xpath = 'data'
ignore_duplicate = True
class item(ItemElement):
klass = MarketOrder
obj_id = Dict('orderId', default=None)
obj_quantity = CleanDecimal.SI(Dict('size'))
obj_date = Date(CleanText(Dict('created')))
obj_state = Title(Dict('status'))
obj__product_id = CleanText(Dict('productId'))
obj_direction = Map(CleanText(Dict('buysell')), MARKET_ORDER_DIRECTIONS, MarketOrderDirection.UNKNOWN)
obj_order_type = Map(Dict('orderTypeId'), MARKET_ORDER_TYPES, MarketOrderType.UNKNOWN)
def obj_ordervalue(self):
if Dict('orderTypeId')(self) != 0:
# Not applicable
return NotAvailable
return CleanDecimal.SI(Dict('price'))(self)
# Some information is not available in this JSON
# so we fetch it in the 'products' dictionary.
# There is no info regarding unitprice, unitvalue & payment method.
def _product(self):
return self.page.browser.get_product(str(Field('_product_id')(self)))
def obj_label(self):
return self._product()['name']
def obj_currency(self):
return Currency().filter(self._product()['currency'])
def obj_code(self):
return IsinCode(default=NotAvailable).filter(self._product()['isin'])
def obj_stock_symbol(self):
return CleanText(default=NotAvailable).filter(self._product().get('symbol'))
def obj_stock_market(self):
exchanges = Env('exchanges')(self)
exchange_id = self._product()['exchangeId']
if exchange_id:
return exchanges.get(int(exchange_id), NotAvailable)
return NotAvailable
def validate(self, obj):
# Some rejected orders do not have an ID, we skip them
return obj.id
class Transaction(FrenchTransaction):
PATTERNS = [(re.compile('^(Deposit|Versement)'), FrenchTransaction.TYPE_DEPOSIT),
(re.compile('^(Buy|Sell|Achat|Vente)'), FrenchTransaction.TYPE_ORDER),
(re.compile(u'^(?P.*)'), FrenchTransaction.TYPE_BANK),
]
class HistoryPage(LoggedPage, JsonPage):
@method
class iter_history(DictElement):
def find_elements(self):
return self.el.get('data', {}).get('cashMovements', [])
class item(ItemElement):
klass = Transaction
def condition(self):
# Transactions without amount are ignored even on the website
return Dict('change', default=None)(self)
obj_raw = Transaction.Raw(CleanText(Dict('description')))
obj_date = Date(CleanText(Dict('date')))
obj__isin = Regexp(Dict('description'), r'\((.{12}?)\)', nth=-1, default=None)
obj__number = Regexp(Dict('description'), r'^([Aa]chat|[Vv]ente|[Bb]uy|[Ss]ell) (\d+[,.]?\d*)', template='\\2', default=None)
obj__datetime = Dict('date')
def obj__action(self):
if not Field('_isin')(self):
return
label = Field('raw')(self).split()[0]
labels = {
'Buy': 'B',
'Achat': 'B',
'Compra': 'B',
'Kauf': 'B',
'Sell': 'S',
'Vente': 'S',
'Venta': 'S',
'Venda': 'S',
'Verkauf': 'S',
'Taxe': None,
'Frais': None,
'Intérêts': None,
'Comisión': None,
'Custo': None,
'Einrichtung': None,
'DEGIRO': None,
'TAKE': None,
'STOCK': None,
'SUBSCRIPTION': None,
'REDEEMED': None,
'ISIN': None,
'MERGER:': None,
'EXPIRATION': None,
'SETTLEMENT': None,
'ASSIGNMENT': None,
'ON': None,
# make sure we don't miss transactions labels specifying an ISIN
}
if label not in labels:
self.logger.warning('Unknown action label: %s', label)
return labels.get(label)
def obj_amount(self):
if Env('account_currency')(self) == Dict('currency')(self):
return float_to_decimal(Dict('change')(self))
# The amount is not displayed so we only retrieve the original_amount
return NotAvailable
def obj_original_amount(self):
if Env('account_currency')(self) == Dict('currency')(self):
return NotAvailable
return float_to_decimal(Dict('change')(self))
def obj_original_currency(self):
if Env('account_currency')(self) == Dict('currency')(self):
return NotAvailable
return Currency(Dict('currency'))(self)
def obj_investments(self):
tr_investment_list = Env('transaction_investments')(self).v
isin = Field('_isin')(self)
action = Field('_action')(self)
if isin and action:
tr_inv_key = (isin, action, Field('_datetime')(self))
try:
return [tr_investment_list[tr_inv_key]]
except KeyError:
pass
return []
def validate(self, obj):
assert not empty(obj.amount) or not empty(obj.original_amount), 'This transaction has no amount and no original_amount!'
return True
@method
class iter_transaction_investments(DictElement):
item_xpath = 'data'
class item(ItemElement):
klass = Investment
obj__product_id = Dict('productId')
obj_quantity = CleanDecimal(Dict('quantity'))
obj_unitvalue = CleanDecimal(Dict('price'))
obj_vdate = Date(CleanText(Dict('date')))
obj__action = Dict('buysell')
obj__datetime = Dict('date')
def _product(self):
return self.page.browser.get_product(str(Field('_product_id')(self)))
def obj_label(self):
return self._product()['name']
def obj_code(self):
code = self._product()['isin']
if is_isin_valid(code):
# Prefix CFD (Contrats for difference) ISIN codes with "XX-"
# to avoid id_security duplicates in the database
if "- CFD" in Field('label')(self):
return "XX-" + code
return code
return NotAvailable
def obj_code_type(self):
if empty(Field('code')(self)):
return NotAvailable
return Investment.CODE_TYPE_ISIN
def get_products(self):
return set(d['productId'] for d in self.doc['data'])
class ExchangesPage(JsonPage):
def get_stock_market_exchanges(self):
exchanges = {}
for exchange in self.doc['exchanges']:
market_code = exchange.get('code')
if market_code:
exchanges[exchange['id']] = market_code
assert exchanges, 'Could not fetch stock market exchanges'
return exchanges
woob-master-5f3d558793b537a74480241ac6981479f5938cd3-modules-degiro/modules/degiro/requirements.txt 0000664 0000000 0000000 00000000014 14575653726 0031513 0 ustar 00root root 0000000 0000000 woob ~= 3.2