15469557513

Committed 05 Jun 2025 02:22PM UTC coverage: 95.087% (-0.7%) from 95.791%

Build # 15469557513

Build Type

Pull #346

github

Committed by

web-flow

Commit Message

Merge 93911324c into 01ae49451

Pull Request Pull Request #346: Add support for INVEXPENSE

Run Details

80 of 87 new or added lines in 4 files covered. (91.95%)

3 existing lines in 3 files now uncovered.

987 of 1038 relevant lines covered (95.09%)

3.8 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.76

/src/ofxstatement/statement.py

"""Statement model"""

from typing import List, Optional
from datetime import datetime
from decimal import Decimal as D
from hashlib import sha1
from pprint import pformat
from math import isclose

from ofxstatement import exceptions

TRANSACTION_TYPES = [
    "CREDIT",  # Generic credit
    "DEBIT",  # Generic debit
    "INT",  # Interest earned or paid
    "DIV",  # Dividend
    "FEE",  # FI fee
    "SRVCHG",  # Service charge
    "DEP",  # Deposit
    "ATM",  # ATM debit or credit
    "POS",  # Point of sale debit or credit
    "XFER",  # Transfer
    "CHECK",  # Check
    "PAYMENT",  # Electronic payment
    "CASH",  # Cash withdrawal
    "DIRECTDEP",  # Direct deposit
    "DIRECTDEBIT",  # Merchant initiated debit
    "REPEATPMT",  # Repeating payment/standing order
    "OTHER",  # Other
]

INVEST_TRANSACTION_TYPES = [
    "BUYDEBT",
    "BUYMF",
    "BUYSTOCK",
    "INCOME",
    "INVEXPENSE",
    "INVBANKTRAN",
    "SELLDEBT",
    "SELLMF",
    "SELLSTOCK",
    "TRANSFER",
]

INVEST_TRANSACTION_BUYTYPES = [
    "BUY",
    "BUYTOCOVER",  # end short sale
]

INVEST_TRANSACTION_SELLTYPES = [
    "SELL",
    "SELLSHORT",  # open short sale
]

INVEST_TRANSACTION_INCOMETYPES = [
    "CGLONG",
    "CGSHORT",
    "DIV",
    "INTEREST",
    "MISC",
]

INVBANKTRAN_TYPES_DETAILED = [
    "INT",
    "XFER",
    "DEBIT",
    "CREDIT",
    "SRVCHG",
    "OTHER",
]

ACCOUNT_TYPE = [
    "CHECKING",  # Checking
    "SAVINGS",  # Savings
    "MONEYMRKT",  # Money Market
    "CREDITLINE",  # Line of credit
]


# Inspired by "How to print instances of a class using print()?"
# on stackoverflow.com
class Printable:
    def __repr__(self) -> str:  # pragma: no cover
        # do not set width to 1 because that makes the output really ugly
        return "<" + type(self).__name__ + "> " + pformat(vars(self), indent=4)


class Statement(Printable):
    """Statement object containing statement items"""

    lines: List["StatementLine"]
    invest_lines: List["InvestStatementLine"]

    currency: Optional[str] = None
    bank_id: Optional[str] = None
    broker_id: Optional[str] = None
    account_id: Optional[str] = None
    # Type of account, must be one of ACCOUNT_TYPE
    account_type: Optional[str] = None

    start_balance: Optional[D] = None
    start_date: Optional[datetime] = None

    end_balance: Optional[D] = None
    end_date: Optional[datetime] = None

    def __init__(
        self,
        bank_id: Optional[str] = None,
        account_id: Optional[str] = None,
        currency: Optional[str] = None,
        account_type: str = "CHECKING",
    ) -> None:
        self.lines = []
        self.invest_lines = []
        self.bank_id = bank_id
        self.account_id = account_id
        self.currency = currency
        self.account_type = account_type

    def assert_valid(self) -> None:  # pragma: no cover
        if not (self.start_balance is None or self.end_balance is None):
            total_amount = sum(
                [sl.amount for sl in self.lines if sl.amount is not None], D(0)
            )

            msg = (
                "Start balance ({0}) plus the total amount ({1}) "
                "should be equal to the end balance ({2})".format(
                    self.start_balance, total_amount, self.end_balance
                )
            )
            if not isclose(self.start_balance + total_amount, self.end_balance):
                raise exceptions.ValidationError(msg, self)


class StatementLine(Printable):
    """Statement line data."""

    id: Optional[str]
    # Date transaction was posted to account
    date: Optional[datetime]
    memo: Optional[str]

    # Amount of transaction
    amount: Optional[D]

    # additional fields
    payee: Optional[str]

    # Date user initiated transaction, if known
    date_user: Optional[datetime]

    # Check (or other reference) number
    check_no: Optional[str]

    # Reference number that uniquely identifies the transaction. Can be used in
    # addition to or instead of a check_no
    refnum: Optional[str]

    # Transaction type, must be one of TRANSACTION_TYPES
    trntype: Optional[str] = "CHECK"

    # Optional BankAccount instance
    bank_account_to: Optional["BankAccount"] = None

    # Currency this line is expressed in (if different from statement
    # currency). Available since 0.7.2
    currency: Optional["Currency"] = None

    # Original amount and the original (foreign) currency. Available since 0.7.2
    orig_currency: Optional["Currency"] = None

    def __init__(
        self,
        id: Optional[str] = None,
        date: Optional[datetime] = None,
        memo: Optional[str] = None,
        amount: Optional[D] = None,
    ) -> None:
        self.id = id
        self.date = date
        self.memo = memo
        self.amount = amount

        self.date_user = None
        self.payee = None
        self.check_no = None
        self.refnum = None

    def __str__(self) -> str:  # pragma: no cover
        return """
        ID: %s, date: %s, amount: %s, payee: %s
        memo: %s
        check no.: %s
        """ % (
            self.id,
            self.date,
            self.amount,
            self.payee,
            self.memo,
            self.check_no,
        )

    def assert_valid(self) -> None:
        """Ensure that fields have valid values"""
        assert (
            self.trntype in TRANSACTION_TYPES
        ), "trntype %s is not valid, must be one of %s" % (
            self.trntype,
            TRANSACTION_TYPES,
        )

        if self.bank_account_to:
            self.bank_account_to.assert_valid()

        assert self.id or self.check_no or self.refnum


class Currency(Printable):
    """CURRENCY and ORIGCURRENCY aggregates from OFX

    See section 5.2 from OFX spec version 2.2.
    """

    # ISO-4217 3-letter currency identifier
    symbol: str
    # Ratio of statement currency to `symbol` currency
    rate: Optional[D]

    def __init__(self, symbol: str, rate: Optional[D] = None) -> None:
        self.symbol = symbol
        self.rate = rate


class InvestStatementLine(Printable):
    """Invest statement line data."""

    id: Optional[str]
    # Date transaction was posted to account
    date: Optional[datetime]
    memo: Optional[str]

    # ID or ticker of underlying security
    security_id: Optional[str]
    # Transaction type, must be one of INVEST_TRANSACTION_TYPES
    trntype: Optional[str]
    # More detailed information about transaction, must be one of INVEST_TRANSACTION_TYPES_DETAILED
    trntype_detailed: Optional[str]

    # Amount of transaction
    amount: Optional[D]
    fees: Optional[D] = None
    unit_price: Optional[D] = None  # required for buy/sell transactions
    units: Optional[D] = None  # required for buy/sell transactions

    def __init__(
        self,
        id: Optional[str] = None,
        date: Optional[datetime] = None,
        memo: Optional[str] = None,
        trntype: Optional[str] = None,
        trntype_detailed: Optional[str] = None,
        security_id: Optional[str] = None,
        amount: Optional[D] = None,
    ) -> None:
        self.id = id
        self.date = date
        self.memo = memo
        self.trntype = trntype
        self.trntype_detailed = trntype_detailed
        self.security_id = security_id
        self.amount = amount

    def __str__(self) -> str:
        return """
            ID: %s, date: %s, trntype: %s, trntype_detailed: %s, security_id: %s, units: %s, unit_price: %s, amount: %s, fees: %s
            memo: %s
            """ % (
            self.id,
            self.date,
            self.trntype,
            self.trntype_detailed,
            self.security_id,
            self.units,
            self.unit_price,
            self.amount,
            self.fees,
            self.memo,
        )

    def assert_valid(self) -> None:
        """Ensure that fields have valid values"""
        # Every transaction needs an ID and date
        assert self.id
        assert self.date

        # Each transaction type has slightly different requirements
        if self.trntype == "BUYDEBT":
            self.assert_valid_buydebt()
        if self.trntype == "BUYMF" or self.trntype == "BUYSTOCK":
            self.assert_valid_buystock()
        elif self.trntype == "INCOME":
            self.assert_valid_income()
        elif self.trntype == "INVBANKTRAN":
            self.assert_valid_invbanktran()
        elif self.trntype == "INVEXPENSE":
            self.assert_valid_invexpense()
        elif self.trntype == "SELLDEBT":
            self.assert_valid_selldebt()
        elif self.trntype == "SELLMF" or self.trntype == "SELLSTOCK":
            self.assert_valid_sellstock()
        elif self.trntype == "TRANSFER":
            self.assert_valid_transfer()
        else:
            raise AssertionError(
                "trntype %s is not valid, must be one of %s"
                % (
                    self.trntype,
                    INVEST_TRANSACTION_TYPES,
                )
            )

    def assert_valid_buydebt(self):
        assert (
            self.trntype_detailed is None
        ), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
        self.assert_valid_invbuy()

    def assert_valid_buystock(self):
        assert (
            self.trntype_detailed in INVEST_TRANSACTION_BUYTYPES
        ), "trntype_detailed %s is not valid, must be one of %s" % (
            self.trntype_detailed,
            INVEST_TRANSACTION_BUYTYPES,
        )
        self.assert_valid_invbuy()

    def assert_valid_income(self):
        assert (
            self.trntype_detailed in INVEST_TRANSACTION_INCOMETYPES
        ), "trntype_detailed %s is not valid, must be one of %s" % (
            self.trntype_detailed,
            INVEST_TRANSACTION_INCOMETYPES,
        )
        assert self.security_id
        assert self.amount

    def assert_valid_invbanktran(self):
        assert (
            self.trntype_detailed in INVBANKTRAN_TYPES_DETAILED
        ), "trntype_detailed %s is not valid for INVBANKTRAN, must be one of %s" % (
            self.trntype_detailed,
            INVBANKTRAN_TYPES_DETAILED,
        )
        assert self.amount

    def assert_valid_invexpense(self):
        assert (
            self.trntype_detailed is None
        ), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
        assert self.security_id
        assert self.amount

    def assert_valid_selldebt(self):
        assert (
            self.trntype_detailed is None
        ), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
        self.assert_valid_invsell()

    def assert_valid_sellstock(self):
        assert (
            self.trntype_detailed in INVEST_TRANSACTION_SELLTYPES
        ), "trntype_detailed %s is not valid, must be one of %s" % (
            self.trntype_detailed,
            INVEST_TRANSACTION_SELLTYPES,
        )
        self.assert_valid_invsell()

    def assert_valid_transfer(self):
        assert (
            self.trntype_detailed is None
        ), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
        assert self.security_id
        assert self.units

    def assert_valid_invbuy(self):
        assert self.security_id
        assert self.units
        assert self.unit_price
        assert self.amount

    def assert_valid_invsell(self):
        assert self.security_id
        assert self.units
        assert self.unit_price
        assert self.amount


class BankAccount(Printable):
    """Structure corresponding to BANKACCTTO and BANKACCTFROM elements from OFX

    Open Financial Exchange uses the Banking Account aggregate to identify an
    account at an FI. The aggregate contains enough information to uniquely
    identify an account for the purposes of statement.
    """

    # Routing and transit number
    bank_id: str
    # Bank identifier for international banks
    branch_id: Optional[str] = None
    # Account number
    acct_id: str
    # Type of account, must be one of ACCOUNT_TYPE
    acct_type: str
    # Checksum for international banks
    acct_key: Optional[str] = None

    def __init__(self, bank_id: str, acct_id: str, acct_type: str = "CHECKING") -> None:
        self.bank_id = bank_id
        self.acct_id = acct_id
        self.acct_type = acct_type

        self.branch_id = None
        self.acct_key = None

    def assert_valid(self) -> None:
        assert self.acct_type in ACCOUNT_TYPE, (
            "acct_type must be one of %s" % ACCOUNT_TYPE
        )


def generate_transaction_id(stmt_line: StatementLine) -> str:
    """Generate pseudo-unique id for given statement line.

    This function can be used in statement parsers when real transaction id is
    not available in source statement.
    """
    h = sha1()
    assert stmt_line.date is not None
    h.update(stmt_line.date.strftime("%Y-%m-%d %H:%M:%S").encode("utf8"))
    if stmt_line.memo is not None:
        h.update(stmt_line.memo.encode("utf8"))
    if stmt_line.amount is not None:
        h.update(str(stmt_line.amount).encode("utf8"))
    return h.hexdigest()


def generate_unique_transaction_id(stmt_line: StatementLine, unique_id_set: set) -> str:
    """
    Generate a unique transaction id.

    A bit of background: the problem with these transaction id's is that
    they do do not only have to be unique, they also have to stay the same
    for the same transaction every time you generate the statement.  So
    generating random ids will not work, even though they will be unique,
    GnuCash or beancount will recognize these transaction as "new" if you
    happen to generate and import the same statement twice or import two
    statements with overlapping periods.

    The function generate_transaction_id() is deterministic, but does not
    necesserily generate an unique id.

    Therefore this function improves on it since you can create a
    really unique id by adding an increment to the generated id (a string)
    and keep on incrementing till it succeeds.

    These are the steps:
    1) supply a unique id set you want to use for checking uniqueness
    2) next you generate an initial id by calling
       generate_transaction_id()
    3) assign the initial id to the current id (id)
    4) increment a counter while the current id is a member of the set and
       add the counter to the initial id and assign that to the current id
    5) add the current id to the unique id set
    6) return a list of the current id and the counter (if not 0)

    The counter is returned in order to enable the caller to modify
    its statement line, for example the memo field.
    """
    # Save the initial id
    id = initial_id = generate_transaction_id(stmt_line)
    counter = 0
    while id in unique_id_set:
        counter += 1
        id = initial_id + str(counter)

    unique_id_set.add(id)
    return id + ("" if counter == 0 else "-" + str(counter))


def recalculate_balance(stmt: Statement) -> None:
    """Recalculate statement starting and ending dates and balances.

    When starting balance is not available, it will be assumed to be 0.

    This function can be used in statement parsers when balance information is
    not available in source statement.
    """

    total_amount = sum([sl.amount for sl in stmt.lines if sl.amount is not None], D(0))

    stmt.start_balance = stmt.start_balance or D(0)
    stmt.end_balance = stmt.start_balance + total_amount
    stmt.start_date = min(sl.date for sl in stmt.lines if sl.date is not None)
    stmt.end_date = max(sl.date for sl in stmt.lines if sl.date is not None)

1	"""Statement model"""
2
3	from typing import List, Optional	4✔
4	from datetime import datetime	4✔
5	from decimal import Decimal as D	4✔
6	from hashlib import sha1	4✔
7	from pprint import pformat	4✔
8	from math import isclose	4✔
9
10	from ofxstatement import exceptions	4✔
11
12	TRANSACTION_TYPES = [	4✔
13	"CREDIT", # Generic credit
14	"DEBIT", # Generic debit
15	"INT", # Interest earned or paid
16	"DIV", # Dividend
17	"FEE", # FI fee
18	"SRVCHG", # Service charge
19	"DEP", # Deposit
20	"ATM", # ATM debit or credit
21	"POS", # Point of sale debit or credit
22	"XFER", # Transfer
23	"CHECK", # Check
24	"PAYMENT", # Electronic payment
25	"CASH", # Cash withdrawal
26	"DIRECTDEP", # Direct deposit
27	"DIRECTDEBIT", # Merchant initiated debit
28	"REPEATPMT", # Repeating payment/standing order
29	"OTHER", # Other
30	]
31
32	INVEST_TRANSACTION_TYPES = [	4✔
33	"BUYDEBT",
34	"BUYMF",
35	"BUYSTOCK",
36	"INCOME",
37	"INVEXPENSE",
38	"INVBANKTRAN",
39	"SELLDEBT",
40	"SELLMF",
41	"SELLSTOCK",
42	"TRANSFER",
43	]
44
45	INVEST_TRANSACTION_BUYTYPES = [	4✔
46	"BUY",
47	"BUYTOCOVER", # end short sale
48	]
49
50	INVEST_TRANSACTION_SELLTYPES = [	4✔
51	"SELL",
52	"SELLSHORT", # open short sale
53	]
54
55	INVEST_TRANSACTION_INCOMETYPES = [	4✔
56	"CGLONG",
57	"CGSHORT",
58	"DIV",
59	"INTEREST",
60	"MISC",
61	]
62
63	INVBANKTRAN_TYPES_DETAILED = [	4✔
64	"INT",
65	"XFER",
66	"DEBIT",
67	"CREDIT",
68	"SRVCHG",
69	"OTHER",
70	]
71
72	ACCOUNT_TYPE = [	4✔
73	"CHECKING", # Checking
74	"SAVINGS", # Savings
75	"MONEYMRKT", # Money Market
76	"CREDITLINE", # Line of credit
77	]
78
79
80	# Inspired by "How to print instances of a class using print()?"
81	# on stackoverflow.com
82	class Printable:	4✔
83	def __repr__(self) -> str: # pragma: no cover
84	# do not set width to 1 because that makes the output really ugly
85	return "<" + type(self).__name__ + "> " + pformat(vars(self), indent=4)
86
87
88	class Statement(Printable):	4✔
89	"""Statement object containing statement items"""
90
91	lines: List["StatementLine"]	4✔
92	invest_lines: List["InvestStatementLine"]	4✔
93
94	currency: Optional[str] = None	4✔
95	bank_id: Optional[str] = None	4✔
96	broker_id: Optional[str] = None	4✔
97	account_id: Optional[str] = None	4✔
98	# Type of account, must be one of ACCOUNT_TYPE
99	account_type: Optional[str] = None	4✔
100
101	start_balance: Optional[D] = None	4✔
102	start_date: Optional[datetime] = None	4✔
103
104	end_balance: Optional[D] = None	4✔
105	end_date: Optional[datetime] = None	4✔
106
107	def __init__(	4✔
108	self,
109	bank_id: Optional[str] = None,
110	account_id: Optional[str] = None,
111	currency: Optional[str] = None,
112	account_type: str = "CHECKING",
113	) -> None:
114	self.lines = []	4✔
115	self.invest_lines = []	4✔
116	self.bank_id = bank_id	4✔
117	self.account_id = account_id	4✔
118	self.currency = currency	4✔
119	self.account_type = account_type	4✔
120
121	def assert_valid(self) -> None: # pragma: no cover
122	if not (self.start_balance is None or self.end_balance is None):
123	total_amount = sum(
124	[sl.amount for sl in self.lines if sl.amount is not None], D(0)
125	)
126
127	msg = (
128	"Start balance ({0}) plus the total amount ({1}) "
129	"should be equal to the end balance ({2})".format(
130	self.start_balance, total_amount, self.end_balance
131	)
132	)
133	if not isclose(self.start_balance + total_amount, self.end_balance):
134	raise exceptions.ValidationError(msg, self)
135
136
137	class StatementLine(Printable):	4✔
138	"""Statement line data."""
139
140	id: Optional[str]	4✔
141	# Date transaction was posted to account
142	date: Optional[datetime]	4✔
143	memo: Optional[str]	4✔
144
145	# Amount of transaction
146	amount: Optional[D]	4✔
147
148	# additional fields
149	payee: Optional[str]	4✔
150
151	# Date user initiated transaction, if known
152	date_user: Optional[datetime]	4✔
153
154	# Check (or other reference) number
155	check_no: Optional[str]	4✔
156
157	# Reference number that uniquely identifies the transaction. Can be used in
158	# addition to or instead of a check_no
159	refnum: Optional[str]	4✔
160
161	# Transaction type, must be one of TRANSACTION_TYPES
162	trntype: Optional[str] = "CHECK"	4✔
163
164	# Optional BankAccount instance
165	bank_account_to: Optional["BankAccount"] = None	4✔
166
167	# Currency this line is expressed in (if different from statement
168	# currency). Available since 0.7.2
169	currency: Optional["Currency"] = None	4✔
170
171	# Original amount and the original (foreign) currency. Available since 0.7.2
172	orig_currency: Optional["Currency"] = None	4✔
173
174	def __init__(	4✔
175	self,
176	id: Optional[str] = None,
177	date: Optional[datetime] = None,
178	memo: Optional[str] = None,
179	amount: Optional[D] = None,
180	) -> None:
181	self.id = id	4✔
182	self.date = date	4✔
183	self.memo = memo	4✔
184	self.amount = amount	4✔
185
186	self.date_user = None	4✔
187	self.payee = None	4✔
188	self.check_no = None	4✔
189	self.refnum = None	4✔
190
191	def __str__(self) -> str: # pragma: no cover
192	return """
193	ID: %s, date: %s, amount: %s, payee: %s
194	memo: %s
195	check no.: %s
196	""" % (
197	self.id,
198	self.date,
199	self.amount,
200	self.payee,
201	self.memo,
202	self.check_no,
203	)
204
205	def assert_valid(self) -> None:	4✔
206	"""Ensure that fields have valid values"""
207	assert (	4✔
208	self.trntype in TRANSACTION_TYPES
209	), "trntype %s is not valid, must be one of %s" % (
210	self.trntype,
211	TRANSACTION_TYPES,
212	)
213
214	if self.bank_account_to:	4✔
215	self.bank_account_to.assert_valid()	×
216
217	assert self.id or self.check_no or self.refnum	4✔
218
219
220	class Currency(Printable):	4✔
221	"""CURRENCY and ORIGCURRENCY aggregates from OFX
222
223	See section 5.2 from OFX spec version 2.2.
224	"""
225
226	# ISO-4217 3-letter currency identifier
227	symbol: str	4✔
228	# Ratio of statement currency to `symbol` currency
229	rate: Optional[D]	4✔
230
231	def __init__(self, symbol: str, rate: Optional[D] = None) -> None:	4✔
232	self.symbol = symbol	4✔
233	self.rate = rate	4✔
234
235
236	class InvestStatementLine(Printable):	4✔
237	"""Invest statement line data."""
238
239	id: Optional[str]	4✔
240	# Date transaction was posted to account
241	date: Optional[datetime]	4✔
242	memo: Optional[str]	4✔
243
244	# ID or ticker of underlying security
245	security_id: Optional[str]	4✔
246	# Transaction type, must be one of INVEST_TRANSACTION_TYPES
247	trntype: Optional[str]	4✔
248	# More detailed information about transaction, must be one of INVEST_TRANSACTION_TYPES_DETAILED
249	trntype_detailed: Optional[str]	4✔
250
251	# Amount of transaction
252	amount: Optional[D]	4✔
253	fees: Optional[D] = None	4✔
254	unit_price: Optional[D] = None # required for buy/sell transactions	4✔
255	units: Optional[D] = None # required for buy/sell transactions	4✔
256
257	def __init__(	4✔
258	self,
259	id: Optional[str] = None,
260	date: Optional[datetime] = None,
261	memo: Optional[str] = None,
262	trntype: Optional[str] = None,
263	trntype_detailed: Optional[str] = None,
264	security_id: Optional[str] = None,
265	amount: Optional[D] = None,
266	) -> None:
267	self.id = id	4✔
268	self.date = date	4✔
269	self.memo = memo	4✔
270	self.trntype = trntype	4✔
271	self.trntype_detailed = trntype_detailed	4✔
272	self.security_id = security_id	4✔
273	self.amount = amount	4✔
274
275	def __str__(self) -> str:	4✔
276	return """	×
277	ID: %s, date: %s, trntype: %s, trntype_detailed: %s, security_id: %s, units: %s, unit_price: %s, amount: %s, fees: %s
278	memo: %s
279	""" % (
280	self.id,
281	self.date,
282	self.trntype,
283	self.trntype_detailed,
284	self.security_id,
285	self.units,
286	self.unit_price,
287	self.amount,
288	self.fees,
289	self.memo,
290	)
291
292	def assert_valid(self) -> None:	4✔
293	"""Ensure that fields have valid values"""
294	# Every transaction needs an ID and date
295	assert self.id	4✔
296	assert self.date	4✔
297
298	# Each transaction type has slightly different requirements
299	if self.trntype == "BUYDEBT":	4✔
NEW 300	self.assert_valid_buydebt()	×
301	if self.trntype == "BUYMF" or self.trntype == "BUYSTOCK":	4✔
302	self.assert_valid_buystock()	4✔
303	elif self.trntype == "INCOME":	4✔
304	self.assert_valid_income()	4✔
305	elif self.trntype == "INVBANKTRAN":	4✔
306	self.assert_valid_invbanktran()	4✔
307	elif self.trntype == "INVEXPENSE":	4✔
308	self.assert_valid_invexpense()	4✔
309	elif self.trntype == "SELLDEBT":	4✔
NEW 310	self.assert_valid_selldebt()	×
311	elif self.trntype == "SELLMF" or self.trntype == "SELLSTOCK":	4✔
312	self.assert_valid_sellstock()	4✔
313	elif self.trntype == "TRANSFER":	4✔
314	self.assert_valid_transfer()	4✔
315	else:
NEW 316	raise AssertionError(	×
317	"trntype %s is not valid, must be one of %s"
318	% (
319	self.trntype,
320	INVEST_TRANSACTION_TYPES,
321	)
322	)
323
324	def assert_valid_buydebt(self):	4✔
NEW 325	assert (	×
326	self.trntype_detailed is None
327	), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
NEW 328	self.assert_valid_invbuy()	×
329
330	def assert_valid_buystock(self):	4✔
331	assert (	4✔
332	self.trntype_detailed in INVEST_TRANSACTION_BUYTYPES
333	), "trntype_detailed %s is not valid, must be one of %s" % (
334	self.trntype_detailed,
335	INVEST_TRANSACTION_BUYTYPES,
336	)
337	self.assert_valid_invbuy()	4✔
338
339	def assert_valid_income(self):	4✔
340	assert (	4✔
341	self.trntype_detailed in INVEST_TRANSACTION_INCOMETYPES
342	), "trntype_detailed %s is not valid, must be one of %s" % (
343	self.trntype_detailed,
344	INVEST_TRANSACTION_INCOMETYPES,
345	)
346	assert self.security_id	4✔
347	assert self.amount	4✔
348
349	def assert_valid_invbanktran(self):	4✔
350	assert (	4✔
351	self.trntype_detailed in INVBANKTRAN_TYPES_DETAILED
352	), "trntype_detailed %s is not valid for INVBANKTRAN, must be one of %s" % (
353	self.trntype_detailed,
354	INVBANKTRAN_TYPES_DETAILED,
355	)
356	assert self.amount	4✔
357
358	def assert_valid_invexpense(self):	4✔
359	assert (	4✔
360	self.trntype_detailed is None
361	), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
362	assert self.security_id	4✔
363	assert self.amount	4✔
364
365	def assert_valid_selldebt(self):	4✔
NEW 366	assert (	×
367	self.trntype_detailed is None
368	), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
NEW 369	self.assert_valid_invsell()	×
370
371	def assert_valid_sellstock(self):	4✔
372	assert (	4✔
373	self.trntype_detailed in INVEST_TRANSACTION_SELLTYPES
374	), "trntype_detailed %s is not valid, must be one of %s" % (
375	self.trntype_detailed,
376	INVEST_TRANSACTION_SELLTYPES,
377	)
378	self.assert_valid_invsell()	4✔
379
380	def assert_valid_transfer(self):	4✔
381	assert (	4✔
382	self.trntype_detailed is None
383	), f"trntype_detailed '{self.trntype_detailed}' should be empty for {self.trntype}"
384	assert self.security_id	4✔
385	assert self.units	4✔
386
387	def assert_valid_invbuy(self):	4✔
388	assert self.security_id	4✔
389	assert self.units	4✔
390	assert self.unit_price	4✔
391	assert self.amount	4✔
392
393	def assert_valid_invsell(self):	4✔
394	assert self.security_id	4✔
395	assert self.units	4✔
396	assert self.unit_price	4✔
397	assert self.amount	4✔
398
399
400	class BankAccount(Printable):	4✔
401	"""Structure corresponding to BANKACCTTO and BANKACCTFROM elements from OFX
402
403	Open Financial Exchange uses the Banking Account aggregate to identify an
404	account at an FI. The aggregate contains enough information to uniquely
405	identify an account for the purposes of statement.
406	"""
407
408	# Routing and transit number
409	bank_id: str	4✔
410	# Bank identifier for international banks
411	branch_id: Optional[str] = None	4✔
412	# Account number
413	acct_id: str	4✔
414	# Type of account, must be one of ACCOUNT_TYPE
415	acct_type: str	4✔
416	# Checksum for international banks
417	acct_key: Optional[str] = None	4✔
418
419	def __init__(self, bank_id: str, acct_id: str, acct_type: str = "CHECKING") -> None:	4✔
420	self.bank_id = bank_id	4✔
421	self.acct_id = acct_id	4✔
422	self.acct_type = acct_type	4✔
423
424	self.branch_id = None	4✔
425	self.acct_key = None	4✔
426
427	def assert_valid(self) -> None:	4✔
428	assert self.acct_type in ACCOUNT_TYPE, (	×
429	"acct_type must be one of %s" % ACCOUNT_TYPE
430	)
431
432
433	def generate_transaction_id(stmt_line: StatementLine) -> str:	4✔
434	"""Generate pseudo-unique id for given statement line.
435
436	This function can be used in statement parsers when real transaction id is
437	not available in source statement.
438	"""
439	h = sha1()	4✔
440	assert stmt_line.date is not None	4✔
441	h.update(stmt_line.date.strftime("%Y-%m-%d %H:%M:%S").encode("utf8"))	4✔
442	if stmt_line.memo is not None:	4✔
443	h.update(stmt_line.memo.encode("utf8"))	4✔
444	if stmt_line.amount is not None:	4✔
445	h.update(str(stmt_line.amount).encode("utf8"))	4✔
446	return h.hexdigest()	4✔
447
448
449	def generate_unique_transaction_id(stmt_line: StatementLine, unique_id_set: set) -> str:	4✔
450	"""
451	Generate a unique transaction id.
452
453	A bit of background: the problem with these transaction id's is that
454	they do do not only have to be unique, they also have to stay the same
455	for the same transaction every time you generate the statement. So
456	generating random ids will not work, even though they will be unique,
457	GnuCash or beancount will recognize these transaction as "new" if you
458	happen to generate and import the same statement twice or import two
459	statements with overlapping periods.
460
461	The function generate_transaction_id() is deterministic, but does not
462	necesserily generate an unique id.
463
464	Therefore this function improves on it since you can create a
465	really unique id by adding an increment to the generated id (a string)
466	and keep on incrementing till it succeeds.
467
468	These are the steps:
469	1) supply a unique id set you want to use for checking uniqueness
470	2) next you generate an initial id by calling
471	generate_transaction_id()
472	3) assign the initial id to the current id (id)
473	4) increment a counter while the current id is a member of the set and
474	add the counter to the initial id and assign that to the current id
475	5) add the current id to the unique id set
476	6) return a list of the current id and the counter (if not 0)
477
478	The counter is returned in order to enable the caller to modify
479	its statement line, for example the memo field.
480	"""
481	# Save the initial id
482	id = initial_id = generate_transaction_id(stmt_line)	4✔
483	counter = 0	4✔
484	while id in unique_id_set:	4✔
485	counter += 1	4✔
486	id = initial_id + str(counter)	4✔
487
488	unique_id_set.add(id)	4✔
489	return id + ("" if counter == 0 else "-" + str(counter))	4✔
490
491
492	def recalculate_balance(stmt: Statement) -> None:	4✔
493	"""Recalculate statement starting and ending dates and balances.
494
495	When starting balance is not available, it will be assumed to be 0.
496
497	This function can be used in statement parsers when balance information is
498	not available in source statement.
499	"""
500
501	total_amount = sum([sl.amount for sl in stmt.lines if sl.amount is not None], D(0))	×
502
503	stmt.start_balance = stmt.start_balance or D(0)	×
504	stmt.end_balance = stmt.start_balance + total_amount	×
505	stmt.start_date = min(sl.date for sl in stmt.lines if sl.date is not None)	×
506	stmt.end_date = max(sl.date for sl in stmt.lines if sl.date is not None)	×

kedder / ofxstatement / 15469557513

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous