admin/tradon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Linkage

Browse Source
This commit is contained in:
laurentbarontini
2026-07-07 15:05:04 +02:00
parent ae39ff2f7d
commit b3b14039e0
3 changed files with 2192 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

553
modules/purchase_trade/stock.py
View File

@@ -1489,6 +1489,559 @@ class ShipmentIn(metaclass=PoolMeta):
if not today:
return ''
return today.strftime('%d-%m-%Y')
@staticmethod
def _report_record_key(record):
record_id = getattr(record, 'id', None)
return record_id if record_id is not None else id(record)
@classmethod
def _report_unique_records(cls, records):
unique = []
seen = set()
for record in records:
if not record:
continue
key = cls._report_record_key(record)
if key in seen:
continue
seen.add(key)
unique.append(record)
return unique
def _get_report_linkage_lots(self):
lots = self._get_report_physical_lots()
if lots:
return lots
return self._report_unique_records(
getattr(move, 'lot', None)
for move in (self.incoming_moves or self.moves or []))
def _get_report_linkage_purchase_lines(self):
return self._report_unique_records(
getattr(lot, 'line', None)
for lot in self._get_report_linkage_lots())
def _get_report_linkage_sale_lines(self):
return self._report_unique_records(
getattr(lot, 'sale_line', None)
for lot in self._get_report_linkage_lots())
def _get_report_linkage_purchase(self):
purchases = self._report_unique_records(
getattr(line, 'purchase', None)
for line in self._get_report_linkage_purchase_lines())
return purchases[0] if purchases else None
def _get_report_linkage_sale(self):
sales = self._report_unique_records(
getattr(line, 'sale', None)
for line in self._get_report_linkage_sale_lines())
return sales[0] if sales else None
@staticmethod
def _report_rec_name(record):
return (
getattr(record, 'rec_name', None)
or getattr(record, 'name', None)
or getattr(record, 'number', None)
or '')
@classmethod
def _report_number(cls, record):
return (
getattr(record, 'number', None)
or getattr(record, 'full_number', None)
or cls._report_rec_name(record)
or '')
@staticmethod
def _report_date(value):
if not value:
return ''
if hasattr(value, 'strftime'):
return value.strftime('%d %b %y')
return str(value)
@staticmethod
def _report_amount(value):
if value in (None, ''):
return ''
return format(Decimal(str(value or 0)).quantize(Decimal('0.001')), ',f')
@staticmethod
def _report_price(value):
if value in (None, ''):
return ''
return format(Decimal(str(value or 0)).quantize(Decimal('0.001')), 'f')
@classmethod
def _report_currency_code(cls, record):
currency = getattr(record, 'currency', None)
return (
getattr(currency, 'code', None)
or getattr(currency, 'name', None)
or getattr(currency, 'symbol', None)
or '')
@classmethod
def _report_unit_symbol(cls, line):
unit = getattr(line, 'unit', None)
return (
getattr(unit, 'symbol', None)
or getattr(unit, 'rec_name', None)
or getattr(unit, 'name', None)
or '')
@classmethod
def _report_line_quantity(cls, line):
return Decimal(str(
getattr(line, 'quantity_theorical', None)
or getattr(line, 'quantity', 0)
or 0))
@classmethod
def _report_line_amount(cls, line):
quantity = cls._report_line_quantity(line)
price = Decimal(str(getattr(line, 'unit_price', 0) or 0))
return quantity * price
def _get_report_linkage_product(self):
lines = (
self._get_report_linkage_purchase_lines()
or self._get_report_linkage_sale_lines())
product = getattr(lines[0], 'product', None) if lines else None
if product:
return product
move = self._get_report_primary_move()
return getattr(move, 'product', None) if move else None
def _get_report_linkage_fees(self):
fees = []
seen = set()
def add_fee(fee):
if not fee:
return
key = self._report_record_key(fee)
if key in seen:
return
seen.add(key)
fees.append(fee)
for fee in getattr(self, 'fees', []) or []:
add_fee(fee)
for line in (
self._get_report_linkage_purchase_lines()
+ self._get_report_linkage_sale_lines()):
for fee in getattr(line, 'fees', []) or []:
add_fee(fee)
return fees
@classmethod
def _report_fee_amount(cls, fee):
get_amount = getattr(fee, 'get_amount', None)
if callable(get_amount):
amount = get_amount()
else:
amount = None
if amount in (None, ''):
price = Decimal(str(getattr(fee, 'price', 0) or 0))
quantity = Decimal(str(getattr(fee, 'quantity', 0) or 0))
mode = getattr(fee, 'mode', None)
amount = price if mode == 'lumpsum' else price * quantity
amount = Decimal(str(amount or 0))
return -amount if getattr(fee, 'p_r', None) == 'pay' else amount
def _get_report_linkage_summary_rows(self):
purchase_lines = self._get_report_linkage_purchase_lines()
sale_lines = self._get_report_linkage_sale_lines()
purchase_total = -sum(
self._report_line_amount(line) for line in purchase_lines)
sale_total = sum(self._report_line_amount(line) for line in sale_lines)
rows = []
if purchase_lines:
rows.append(('Purchases', '', purchase_total, purchase_total))
if sale_lines:
rows.append(('Sales', '', sale_total, sale_total))
fee_total = Decimal('0')
for fee in self._get_report_linkage_fees():
product = getattr(fee, 'product', None)
label = self._report_rec_name(product) or 'Fee'
amount = self._report_fee_amount(fee)
fee_total += amount
rows.append(('Costs', label, amount, amount))
if rows:
rows.append(('Costs', 'Total', fee_total, fee_total))
pnl = purchase_total + sale_total + fee_total
rows.append(('P&L', '', pnl, pnl))
return rows
def _get_report_linkage_movement_rows(self):
rows = []
purchase_lines = self._get_report_linkage_purchase_lines()
sale_lines = self._get_report_linkage_sale_lines()
if purchase_lines:
rows.append(('Purchase', '', '', '', '', '', '', '', '', ''))
for line in purchase_lines:
purchase = getattr(line, 'purchase', None)
rows.append(self._get_report_linkage_movement_row(
'Deal', purchase, line))
if sale_lines:
rows.append(('Sale', '', '', '', '', '', '', '', '', ''))
for line in sale_lines:
sale = getattr(line, 'sale', None)
rows.append(self._get_report_linkage_movement_row(
'Deal', sale, line))
if purchase_lines or sale_lines:
balance = (
sum(self._report_line_quantity(line) for line in purchase_lines)
- sum(self._report_line_quantity(line) for line in sale_lines))
unit = self._report_unit_symbol(
purchase_lines[0] if purchase_lines else sale_lines[0])
rows.append((
'Balance (Gain/Loss)', '', '', '',
' '.join(part for part in [
self._format_report_quantity(balance),
unit.upper() if unit else ''] if part),
'', '', '', '', ''))
return rows
def _get_report_linkage_movement_row(self, kind, trade, line):
product = getattr(line, 'product', None)
quantity = ' '.join(part for part in [
self._format_report_quantity(self._report_line_quantity(line)),
(self._report_unit_symbol(line) or '').upper()] if part)
incoterm = getattr(trade, 'incoterm', None)
period = getattr(line, 'period_at', None) or ''
if period == 'laycan':
period = 'Laycan'
return (
kind,
self._report_number(trade),
self._report_rec_name(getattr(trade, 'party', None)),
self._report_rec_name(product),
quantity,
getattr(incoterm, 'code', None) or '',
self._report_rec_name(getattr(trade, 'to_location', None)),
period,
self._report_date(getattr(line, 'from_del', None)),
self._report_date(getattr(line, 'to_del', None)),
)
def _get_report_linkage_pricing_rows(self):
rows = []
for label, lines in (
('Buy', self._get_report_linkage_purchase_lines()),
('Sell', self._get_report_linkage_sale_lines())):
for line in lines:
trade = (
getattr(line, 'purchase', None)
or getattr(line, 'sale', None))
currency = self._report_currency_code(trade)
unit = self._report_unit_symbol(line).upper()
price = self._report_price(getattr(line, 'unit_price', 0))
rows.append((
self._report_number(trade),
label,
(getattr(line, 'price_type', None) or 'Fixed').title(),
'',
self._report_rec_name(getattr(line, 'del_period', None)),
' '.join(part for part in [
price,
'/'.join(part for part in [currency, unit] if part)]
if part),
))
return rows
def _get_report_linkage_detail_rows(self):
rows = []
for line in self._get_report_linkage_sale_lines():
sale = getattr(line, 'sale', None)
rows.append(self._get_report_linkage_detail_row(
'A', 'Deal revenue', sale, line, self._report_line_amount(line)))
for line in self._get_report_linkage_purchase_lines():
purchase = getattr(line, 'purchase', None)
rows.append(self._get_report_linkage_detail_row(
'A', 'Deal expense', purchase, line,
-self._report_line_amount(line)))
for fee in self._get_report_linkage_fees():
rows.append(self._get_report_linkage_fee_detail_row(fee))
total = sum((row[-1] for row in rows), Decimal('0'))
if rows:
rows.append(('A', 'Total', '', '', '', '', total, total))
rows.append(('P&L', '', '', '', '', '', total, total))
return rows
def _get_report_linkage_detail_row(self, group, label, trade, line, amount):
currency = self._report_currency_code(trade)
unit = self._report_unit_symbol(line).upper()
unit_price = self._report_price(getattr(line, 'unit_price', 0))
quantity = self._format_report_quantity(self._report_line_quantity(line))
return (
group,
label,
self._report_rec_name(getattr(trade, 'party', None)),
self._report_number(trade),
' '.join(part for part in [
unit_price,
'/'.join(part for part in [currency, unit] if part)]
if part),
quantity,
amount,
amount,
)
def _get_report_linkage_fee_detail_row(self, fee):
product = getattr(fee, 'product', None)
trade_line = (
getattr(fee, 'line', None)
or getattr(fee, 'sale_line', None)
or (self._get_report_linkage_purchase_lines()
or self._get_report_linkage_sale_lines()
or [None])[0])
trade = (
getattr(trade_line, 'purchase', None)
or getattr(trade_line, 'sale', None))
currency = self._report_currency_code(fee) or self._report_currency_code(trade)
unit = self._get_report_unit_text(getattr(fee, 'unit', None))
amount = self._report_fee_amount(fee)
quantity = getattr(fee, 'quantity', None)
if quantity in (None, '') and trade_line:
quantity = self._report_line_quantity(trade_line)
return (
'A',
self._report_rec_name(product) or 'Fee',
self._report_rec_name(getattr(fee, 'supplier', None)),
self._report_number(trade),
' '.join(part for part in [
self._report_price(getattr(fee, 'price', 0)),
'/'.join(part for part in [currency, unit.upper() if unit else ''] if part)]
if part),
self._format_report_quantity(quantity or 0),
amount,
amount,
)
@classmethod
def _report_column(cls, rows, index, amount=False):
values = []
for row in rows:
value = row[index]
if amount:
value = cls._report_amount(value)
values.append(str(value or ''))
return '\n'.join(values)
@property
def report_linkage_title(self):
purchase = self._get_report_linkage_purchase()
sale = self._get_report_linkage_sale()
reference = (
getattr(purchase, 'reference', None)
or getattr(sale, 'reference', None)
or getattr(self, 'reference', None)
or '')
numbers = '/'.join(filter(None, [
self._report_number(purchase),
self._report_number(sale),
]))
return ' '.join(part for part in ['Linkage', numbers, reference] if part)
@property
def report_linkage_desk(self):
product = self._get_report_linkage_product()
return self._report_rec_name(product)
@property
def report_linkage_book(self):
product = self._get_report_linkage_product()
code = getattr(product, 'code', None) if product else ''
period = self.report_linkage_strategy
return ' '.join(part for part in [code, period] if part)
@property
def report_linkage_strategy(self):
for line in (
self._get_report_linkage_purchase_lines()
+ self._get_report_linkage_sale_lines()):
strategies = list(getattr(line, 'mtm', []) or [])
if strategies:
strategy = getattr(strategies[0], 'strategy', None) or strategies[0]
return self._report_rec_name(strategy)
period = getattr(line, 'del_period', None)
if period:
return self._report_rec_name(period)
return ''
@property
def report_linkage_bl_date(self):
return self.bl_date and self.bl_date.strftime('%A, %B %d, %Y') or ''
@property
def report_linkage_trading_unit(self):
sale = self._get_report_linkage_sale()
party = getattr(sale, 'party', None) if sale else None
return self._report_rec_name(party)
@property
def report_linkage_finance_user(self):
trade = self._get_report_linkage_purchase() or self._get_report_linkage_sale()
operator = getattr(trade, 'operator', None) if trade else None
return self._report_rec_name(operator)
@property
def report_linkage_summary_groups(self):
return self._report_column(self._get_report_linkage_summary_rows(), 0)
@property
def report_linkage_summary_cost_types(self):
return self._report_column(self._get_report_linkage_summary_rows(), 1)
@property
def report_linkage_summary_estimated(self):
return self._report_column(
self._get_report_linkage_summary_rows(), 2, amount=True)
@property
def report_linkage_summary_validated(self):
return self._report_column(
self._get_report_linkage_summary_rows(), 3, amount=True)
@property
def report_linkage_movement_types(self):
return self._report_column(self._get_report_linkage_movement_rows(), 0)
@property
def report_linkage_movement_references(self):
return self._report_column(self._get_report_linkage_movement_rows(), 1)
@property
def report_linkage_movement_counterparts(self):
return self._report_column(self._get_report_linkage_movement_rows(), 2)
@property
def report_linkage_movement_commodities(self):
return self._report_column(self._get_report_linkage_movement_rows(), 3)
@property
def report_linkage_movement_quantities(self):
return self._report_column(self._get_report_linkage_movement_rows(), 4)
@property
def report_linkage_movement_deliveries(self):
return self._report_column(self._get_report_linkage_movement_rows(), 5)
@property
def report_linkage_movement_basis(self):
return self._report_column(self._get_report_linkage_movement_rows(), 6)
@property
def report_linkage_movement_periods(self):
return self._report_column(self._get_report_linkage_movement_rows(), 7)
@property
def report_linkage_movement_from_dates(self):
return self._report_column(self._get_report_linkage_movement_rows(), 8)
@property
def report_linkage_movement_to_dates(self):
return self._report_column(self._get_report_linkage_movement_rows(), 9)
@property
def report_linkage_bank_deal(self):
purchase = self._get_report_linkage_purchase()
return self._report_number(purchase)
@property
def report_linkage_bank_type(self):
return 'Buy' if self._get_report_linkage_purchase() else ''
@property
def report_linkage_bank_name(self):
sale = self._get_report_linkage_sale()
bank = getattr(sale, 'our_bank_account', None) if sale else None
bank_party = getattr(bank, 'bank', None) if bank else None
return self._report_rec_name(bank_party)
@property
def report_linkage_lc_type(self):
return 'N/A'
@property
def report_linkage_lc_number(self):
sale = self._get_report_linkage_sale()
return getattr(sale, 'lc_number', '') or ''
@property
def report_linkage_lc_amount(self):
return ''
@property
def report_linkage_lc_expiry_date(self):
sale = self._get_report_linkage_sale()
return self._report_date(getattr(sale, 'lc_date', None))
@property
def report_linkage_pricing_deals(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 0)
@property
def report_linkage_pricing_sides(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 1)
@property
def report_linkage_pricing_types(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 2)
@property
def report_linkage_pricing_symbols(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 3)
@property
def report_linkage_pricing_periods(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 4)
@property
def report_linkage_pricing_input_prices(self):
return self._report_column(self._get_report_linkage_pricing_rows(), 5)
@property
def report_linkage_detail_groups(self):
return self._report_column(self._get_report_linkage_detail_rows(), 0)
@property
def report_linkage_detail_cost_types(self):
return self._report_column(self._get_report_linkage_detail_rows(), 1)
@property
def report_linkage_detail_counterparts(self):
return self._report_column(self._get_report_linkage_detail_rows(), 2)
@property
def report_linkage_detail_sources(self):
return self._report_column(self._get_report_linkage_detail_rows(), 3)
@property
def report_linkage_detail_unit_prices(self):
return self._report_column(self._get_report_linkage_detail_rows(), 4)
@property
def report_linkage_detail_quantities(self):
return self._report_column(self._get_report_linkage_detail_rows(), 5)
@property
def report_linkage_detail_estimated(self):
return self._report_column(
self._get_report_linkage_detail_rows(), 6, amount=True)
@property
def report_linkage_detail_validated(self):
return self._report_column(
self._get_report_linkage_detail_rows(), 7, amount=True)
def get_rec_name(self, name=None):
if self.number:

86
modules/purchase_trade/tests/test_module.py
View File

@@ -6407,6 +6407,92 @@ class PurchaseTradeTestCase(ModuleTestCase):
self.assertEqual(
shipment.report_insurance_amount, 'USD 550.00')
def test_shipment_linkage_helpers_use_matched_purchase_sale_lot(self):
'linkage helpers expose matched purchase and sale values'
ShipmentIn = Pool().get('stock.shipment.in')
shipment = ShipmentIn()
shipment.bl_date = datetime.date(2026, 4, 30)
shipment.fees = []
shipment.moves = []
shipment.lotqt = []
currency = SimpleNamespace(code='USD')
unit = SimpleNamespace(symbol='MT')
product = SimpleNamespace(
rec_name='Sulphuric Acid',
name='Sulphuric Acid',
code='H2SO4')
purchase = SimpleNamespace(
id=1,
number='P-10',
reference='26.0001',
currency=currency,
party=SimpleNamespace(rec_name='SUPPLIER SA'),
to_location=SimpleNamespace(rec_name='Odda'),
incoterm=SimpleNamespace(code='FOB'),
operator=SimpleNamespace(rec_name='Operator A'))
sale = SimpleNamespace(
id=2,
number='S-20',
reference='26.0001',
currency=currency,
party=SimpleNamespace(rec_name='CLIENT SA'),
to_location=SimpleNamespace(rec_name='Mejillones'),
incoterm=SimpleNamespace(code='CFR'))
period = SimpleNamespace(rec_name='Chile FY26')
purchase_line = SimpleNamespace(
id=11,
product=product,
purchase=purchase,
quantity_theorical=Decimal('100'),
quantity=Decimal('100'),
unit_price=Decimal('80'),
unit=unit,
price_type='fixed',
del_period=period,
period_at='laycan',
from_del=datetime.date(2026, 4, 30),
to_del=datetime.date(2026, 5, 8),
fees=[],
mtm=[])
sale_line = SimpleNamespace(
id=12,
product=product,
sale=sale,
quantity_theorical=Decimal('100'),
quantity=Decimal('100'),
unit_price=Decimal('100'),
unit=unit,
price_type='fixed',
del_period=period,
period_at='laycan',
from_del=datetime.date(2026, 4, 30),
to_del=datetime.date(2026, 4, 30),
fees=[],
mtm=[])
lot = SimpleNamespace(
id=101,
lot_type='physic',
line=purchase_line,
sale_line=sale_line)
shipment.incoming_moves = [SimpleNamespace(lot=lot)]
self.assertEqual(
shipment.report_linkage_title, 'Linkage P-10/S-20 26.0001')
self.assertEqual(shipment.report_linkage_desk, 'Sulphuric Acid')
self.assertEqual(shipment.report_linkage_book, 'H2SO4 Chile FY26')
self.assertEqual(
shipment.report_linkage_bl_date, 'Thursday, April 30, 2026')
self.assertIn('Purchases', shipment.report_linkage_summary_groups)
self.assertIn('Sales', shipment.report_linkage_summary_groups)
self.assertIn('-8,000.000', shipment.report_linkage_summary_estimated)
self.assertIn('10,000.000', shipment.report_linkage_summary_estimated)
self.assertIn('P-10', shipment.report_linkage_movement_references)
self.assertIn('S-20', shipment.report_linkage_movement_references)
self.assertIn('100 MT', shipment.report_linkage_movement_quantities)
self.assertIn('80.000 USD/MT', shipment.report_linkage_pricing_input_prices)
self.assertIn('100.000 USD/MT', shipment.report_linkage_pricing_input_prices)
def test_shipment_insurance_contact_surveyor_prefers_shipment_surveyor(self):
'insurance contact surveyor property uses shipment surveyor first'
ShipmentIn = Pool().get('stock.shipment.in')

1553
modules/stock/linkage.fodt Normal file
View File

File diff suppressed because one or more lines are too long