I tried to create a plugin myself but ended up just creating a script that
created the transfers. It works, but you have to make sure you add them in
order or you'll have issues. It's helpful if you have a lot of lots and
transfer cryptocurrencies between accounts or exchanges...
```
#!/usr/bin/env python3
"""
Beancount Crypto Transfer Transaction Generator v2
This script uses the native Beancount Python API instead of subprocess calls
to query your Beancount file and generate transfer transactions
automatically
using FIFO lot selection.
Usage:
python transfer_generator_v2.py ledger.beancount
"Assets:CryptoExchange" "Assets:PrivateWallet" 20 BSV "Transfer to cold
storage"
"""
import sys
import os
from decimal import Decimal
from datetime import datetime
import argparse
from beancount import loader
from beanquery.query import run_query
def get_account_lots(beancount_file, account, currency):
"""Get all lots for a specific account and currency using native
Beancount API"""
try:
# Load the beancount file
entries, errors, options_map = loader.load_file(beancount_file)
if errors:
print(f"Warning: Found {len(errors)} errors in Beancount file:")
for error in errors[:5]: # Show first 5 errors
print(f" {error}")
if len(errors) > 5:
print(f" ... and {len(errors) - 5} more errors")
# Run the query directly using beanquery
query = f"""
SELECT account,
units(sum(position)) as units,
cost_number as cost,
cost_currency as cost_currency,
cost_date as acquisition_date,
cost_label as lot_label
WHERE account = '{account}' AND currency = '{currency}'
GROUP BY account, cost_number, cost_currency, cost_date, cost_label
ORDER BY cost_date
"""
# This returns structured data, not CSV text!
result_types, result_rows = run_query(entries, options_map, query)
lots = []
for row in result_rows:
# Direct access to typed data instead of parsing CSV strings
units_inventory = row[1] # This is an Inventory object, not
Amount
# Skip empty inventories
if units_inventory is None:
continue
# Handle Beancount Inventory objects
try:
if hasattr(units_inventory, 'is_empty') and units_inventory
.is_empty():
continue
# Extract the Amount from the Inventory
# An Inventory can contain multiple positions, we want the
one for our currency
amount = None
if hasattr(units_inventory, '__iter__'):
for position in units_inventory:
if hasattr(position, 'units') and
position.units.currency
== currency:
amount = position.units
break
if amount is None or amount.number <= 0:
continue
lot = {
'account': str(row[0]) if row[0] else '',
'units': amount.number,
'cost': row[2] if row[2] is not None else None,
'cost_currency': str(row[3]) if row[3] else None,
'acquisition_date': str(row[4]) if row[4] else None,
'lot_label': str(row[5]) if row[5] else None
}
lots.append(lot)
except Exception as e:
print(f"Debug: Exception processing row: {e}")
continue
return lots
except Exception as e:
print(f"Error querying lots: {e}")
return []
def select_lots_fifo(lots, transfer_amount):
"""Select lots using FIFO method for the transfer amount"""
selected_lots = []
remaining = Decimal(str(transfer_amount))
# Sort by acquisition date (FIFO) - handle None dates
sorted_lots = sorted(lots, key=lambda x: x['acquisition_date'] or
'1900-01-01')
for lot in sorted_lots:
if remaining <= 0:
break
# Take either the full lot or remaining amount
take_amount = min(remaining, lot['units'])
selected_lot = {
'account': lot['account'],
'units': take_amount,
'cost': lot['cost'],
'cost_currency': lot['cost_currency'],
'acquisition_date': lot['acquisition_date'],
'lot_label': lot['lot_label']
}
selected_lots.append(selected_lot)
remaining -= take_amount
if remaining > 0:
print(f"Warning: Not enough units available. Short by {remaining}")
return selected_lots
def format_cost_spec(lot):
"""Format the cost specification for a lot"""
parts = []
if lot['cost'] and lot['cost_currency']:
parts.append(f"{lot['cost']} {lot['cost_currency']}")
if lot['acquisition_date']:
parts.append(lot['acquisition_date'])
if lot['lot_label']:
parts.append(f'"{lot["lot_label"]}"')
if parts:
return "{" + ", ".join(parts) + "}"
else:
return "{}"
def generate_transfer_transaction(date, from_account, to_account, currency,
description, selected_lots):
"""Generate a Beancount transfer transaction"""
lines = [f'{date} ! "{description}"']
for i, lot in enumerate(selected_lots):
cost_spec = format_cost_spec(lot)
# Format with proper spacing
from_line = f" {from_account:<45} -{lot['units']} {currency} {
cost_spec}"
to_line = f" {to_account:<45} {lot['units']} {currency} {cost_spec}
"
lines.append(from_line)
lines.append(to_line)
return '\n'.join(lines)
def main():
parser = argparse.ArgumentParser(
description='Generate Beancount crypto transfer transactions with
automatic lot selection (v2 - Native API)'
)
parser.add_argument('beancount_file', help='Path to your Beancount file'
)
parser.add_argument('from_account', help='Source account (e.g.,
Assets:CryptoExchange)')
parser.add_argument('to_account', help='Destination account (e.g.,
Assets:PrivateWallet)')
parser.add_argument('amount', type=str, help='Amount to transfer')
parser.add_argument('currency', help='Currency symbol (e.g., BSV, BTC)')
parser.add_argument('description', help='Transaction description')
parser.add_argument('--date', help='Transaction date (YYYY-MM-DD)',
default=datetime.now().strftime('%Y-%m-%d'))
parser.add_argument('--method', choices=['FIFO', 'LIFO'], default='FIFO'
,
help='Lot selection method (default: FIFO)')
args = parser.parse_args()
# Convert relative path to absolute path
args.beancount_file = os.path.abspath(args.beancount_file)
# Check if file exists
if not os.path.exists(args.beancount_file):
print(f"Error: Beancount file '{args.beancount_file}' does not
exist")
print(f"Current directory: {os.getcwd()}")
print("Available files:")
for f in os.listdir('.'):
if f.endswith('.beancount') or f.endswith('.bean'):
print(f" {f}")
return
# Get lots from source account
print(f"Querying lots in {args.from_account} for {args.currency}...")
lots = get_account_lots(args.beancount_file, args.from_account, args
.currency)
if not lots:
print(f"No lots found in {args.from_account} for {args.currency}")
return
print(f"Found {len(lots)} lots:")
total_available = sum(lot['units'] for lot in lots)
print(f"Total available: {total_available} {args.currency}")
for i, lot in enumerate(lots):
label_display = f' ("{lot["lot_label"]}")' if lot['lot_label'] else
''
print(f" Lot {i+1}: {lot['units']} {args.currency} @ {lot['cost']}
{lot['cost_currency']} ({lot['acquisition_date']}){label_display}")
# Check if we have enough
transfer_amount = Decimal(args.amount)
if transfer_amount > total_available:
print(f"Error: Requested {transfer_amount} {args.currency} but only
{total_available} available")
return
# Select lots using specified method
if args.method == 'LIFO':
lots.reverse() # Reverse for LIFO
selected_lots = select_lots_fifo(lots, transfer_amount)
print(f"\nSelected lots for transfer ({args.method}):")
for i, lot in enumerate(selected_lots):
label_display = f' ("{lot["lot_label"]}")' if lot['lot_label'] else
''
print(f" {lot['units']} {args.currency} @ {lot['cost']} {lot[
'cost_currency']} ({lot['acquisition_date']}){label_display}")
# Generate transaction
transaction = generate_transfer_transaction(
args.date, args.from_account, args.to_account,
args.currency, args.description, selected_lots
)
print(f"\nGenerated transaction:")
print("=" * 80)
print(transaction)
print("=" * 80)
# Save to file option
save = input("\nSave to file? (y/N): ").lower().strip()
if save == 'y':
filename = f"transfer_{args.currency}_{args.date.replace('-', '')}
.beancount"
with open(filename, 'w') as f:
f.write(transaction + '\n')
print(f"Saved to {filename}")
if __name__ == '__main__':
if len(sys.argv) == 1:
# Example usage
print("Example usage:")
print('python transfer_generator_v2.py ledger.beancount
"Assets:CryptoExchange" "Assets:PrivateWallet" 20 BSV "Transfer to cold
storage"')
print("")
print("Optional arguments:")
print(" --date YYYY-MM-DD (default: today)")
print(" --method FIFO|LIFO (default: FIFO)")
sys.exit(1)
main()
```
It works, but there's still a bunch of manual work when I use it. If there
is a more elegant way to do this I would love to hear it.
-Chris
On Friday, June 27, 2025 at 7:22:46 AM UTC-4 [email protected] wrote:
> On Saturday, June 21, 2025 at 3:43:06 PM UTC+9:30 Red S wrote:
>
> Here’s a plugin
> <https://github.com/hoostus/beancount-asset-transfer-plugin> that does
> what you want. But more interestingly, the README on that page explains and
> shows you how to do it.
>
>
> Note that that plugin doesn't actually work due to beancount's loader
> running interpolation on incomplete booking entries before it runs
> transformation plugins. (See bug #1 in the project for more detail.)
>
--
You received this message because you are subscribed to the Google Groups
"Beancount" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to [email protected].
To view this discussion visit
https://groups.google.com/d/msgid/beancount/4bc91995-d3f2-40d0-b5b6-fa36ddb5c5cbn%40googlegroups.com.
