Média de posição no Beancount

  
#!/usr/bin/env python3
from beancount import loader
from beancount.core import data
from beancount.parser import printer
from pathlib import Path
import sys

FIXES_FILE = Path(__file__).parent.parent / "fixes.beancount"
MAIN_FILE = Path(__file__).parent.parent / "main.beancount"

fixers = {}
def fixer(func):
    fixers[func.__name__] = func
    return func
       

def counter(initial=0):
    i = initial
    while True:
        yield i
        i += 1
            

@fixer
def b3(entries):
    from collections import defaultdict
    import datetime
    from decimal import Decimal

    initial_date = datetime.date(1970, 1, 1)

    class Price():
        def __init__(self, amount, price, date):
            self.amount = amount
            if not isinstance(self.amount, Decimal):
                self.amount = Decimal(self.amount)
            self.price = price
            if not isinstance(self.price, Decimal):
                self.price = Decimal(self.price)
            assert isinstance(date, datetime.date)
            self.date = date

        def value(self, amount):
            if not isinstance(amount, Decimal):
                amount = Decimal(amount)
            return self.price * amount

        def __repr__(self):
            return f"Price(amount={self.amount}, price={self.price} day={self.date})"

    class AccountState():
        def __init__(self, currency):
            self.currency = currency
            self.price = defaultdict(lambda: Price(0, 0, initial_date))

        def handle_posting(self, entry, account, amount, price, share_currency):
            if self.price[share_currency].date == initial_date:
                self.price[share_currency].date = entry.date
                self.price[share_currency].amount = amount
                self.price[share_currency].price = price
                print(self.price[share_currency], account, amount, price, share_currency, file=sys.stderr)
                return None
            if amount == 0:
                return None
            default_args = dict(
                narration="Equilíbrio de preço médio",
                date=entry.date,
                meta=entry.meta,
                flag='*',
                tags=set(),
                links=set(),
            )
            if amount < 0:
                self.price[share_currency].amount += amount # - ia somar
                print(self.price[share_currency], account, amount, price, share_currency, file=sys.stderr)
                return
            final_cost = (self.price[share_currency].amount * self.price[share_currency].price) + (amount * price)
            final_amount = (self.price[share_currency].amount + amount)
            final_avgcost = final_cost / final_amount
            ret = dict(
                **default_args,
                postings=[
                    dict(
                        account=account,
                        units=data.Amount(-self.price[share_currency].amount, share_currency),
                        cost=data.Cost(self.price[share_currency].price, self.currency, self.price[share_currency].date, None),
                    ),
                    dict(
                        account=account,
                        units=data.Amount(-amount, share_currency),
                        cost=data.Cost(price, self.currency, entry.date, None),
                    ),
                    dict(
                        account=account,
                        units=data.Amount(final_amount, share_currency),
                        cost=data.Cost(final_avgcost, self.currency, entry.date, None),
                    )
                ]
            )
            ret['postings'] = [posting for posting in ret['postings'] if posting['units'].number != 0]
            self.price[share_currency].date = entry.date
            self.price[share_currency].amount = final_amount
            self.price[share_currency].price = final_avgcost
            print(self.price[share_currency], account, amount, price, share_currency, file=sys.stderr)
            return ret
    errors = []
    extra_entries = []
    accounts = []
    currencies = {}
    line_counter = counter(initial=1)
    price = {}

    yield dict(
        narration="Teste",
        date=datetime.date.today(),
        postings=[]
    )

    for entry in entries:
        if isinstance(entry, data.Open):
            if entry.meta.get('b3_automations') == "TRUE":
                price[entry.account] = AccountState(
                    currency = entry.meta.get('b3_currency', 'BRL')
                )
            continue
        # print(entry)
        if isinstance(entry, data.Transaction):
            date = entry.date
            for posting in entry.postings:
                amount = posting.units.number
                currency = posting.units.currency
                if posting.cost is None:
                    continue
                cost = posting.cost.number
                cost_currency = posting.cost.currency
                account = posting.account
                # print(list(price.keys()))
                if account not in list(price.keys()):
                    continue

                if cost_currency != price[account].currency:
                    errors.append(f"in {entry.meta.filename}:{entry.meta.fileno}: invalid currency: expected: {cost_currency} got: {price[entry.account].currency}")
                    continue
                new_entry = price[account].handle_posting(
                    entry=entry,
                    account=account,
                    amount=amount,
                    price=cost,
                    share_currency=currency,
                )
                if new_entry is None:
                    continue
                yield new_entry


if __name__ == '__main__':
    default_transaction = dict(
        meta=None,
        flag="*",
        payee="BEANCOUNT",
        narration="",
        tags=set(),
        links=set(),
        postings=[]
    )
    default_posting = dict(
        price=None,
        flag=None,
        meta=None,
    )
    entries, _, _ = loader.load_file(Path(__file__).parent.parent / "main.beancount")
    entries = [entry for entry in entries if not (isinstance(entry, data.Transaction) and entry.payee == "BEANCOUNT")]
    with FIXES_FILE.open('w') as f:
        print(';; Generated using scripts/codegen_fixes.py. Do not edit.', file=f)
        print(file=f)
        for fixer_name, fixer in fixers.items():
            for item in fixer(entries):
                processed_item = data.Transaction(**{
                    **default_transaction,
                    **item,
                    'postings': [data.Posting(**{**default_posting, **posting}) for posting in item.get('postings', [])]
                })
                printer.print_entry(processed_item, file=f)

from beancount.core import data
from datetime import date
from pathlib import Path
import sys

__plugins__ = ["autodocs"]

DOCS_FOLDER=Path(__file__).parent.parent / "docs"

def autodocs(entries, options_map):
    extra_entries = []
    for entry in entries:
        if not isinstance(entry, data.Transaction):
            continue
        # print(entry)
        doc_name = entry.meta.get('doc_name')
        if doc_name is None:
            continue
        doc_account = entry.meta.get('doc_account')
        if doc_account is None:
            if len(entry.postings) > 0:
                doc_account=entry.postings[0].account
        if doc_account is None:
            continue
        tx_date = entry.date
        to_append = data.Document(
            meta=data.new_metadata(entry.meta['filename'], entry.meta['lineno']),
            account=doc_account,
            filename=str(DOCS_FOLDER / f"{tx_date.year}{tx_date.month:02}{tx_date.day:02}-{doc_name}"),
            date=tx_date,
            tags=set(),
            links=set(),
        )
        # print(entry, file=sys.stderr)
        # print(to_append, file=sys.stderr)
        extra_entries.append(to_append)
    return [*entries, *extra_entries], []