1331 lines
50 KiB
Python
1331 lines
50 KiB
Python
from trytond.model import fields
|
|
from trytond.report import Report
|
|
from trytond.pool import Pool, PoolMeta
|
|
from trytond.pyson import Bool, Eval, Id, If, PYSONEncoder
|
|
from trytond.model import (ModelSQL, ModelView)
|
|
from trytond.tools import is_full_text, lstrip_wildcard
|
|
from trytond.transaction import Transaction, inactive_records
|
|
from decimal import getcontext, Decimal, ROUND_HALF_UP
|
|
from sql.aggregate import Count, Max, Min, Sum, Avg, BoolOr
|
|
from sql.conditionals import Case
|
|
from sql import Column, Literal
|
|
from sql.functions import CurrentTimestamp, DateTrunc
|
|
from trytond.wizard import Button, StateAction, StateTransition, StateView, Wizard
|
|
from itertools import chain, groupby
|
|
from operator import itemgetter
|
|
import datetime
|
|
import logging
|
|
import re
|
|
from collections import defaultdict
|
|
from trytond.exceptions import UserWarning, UserError
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
VALTYPE = [
|
|
('priced', 'Price'),
|
|
('pur. priced', 'Pur. price'),
|
|
('pur. efp', 'Pur. efp'),
|
|
('sale priced', 'Sale price'),
|
|
('sale efp', 'Sale efp'),
|
|
('line fee', 'Line fee'),
|
|
('pur. fee', 'Pur. fee'),
|
|
('sale fee', 'Sale fee'),
|
|
('shipment fee', 'Shipment fee'),
|
|
('market', 'Market'),
|
|
('derivative', 'Derivative'),
|
|
]
|
|
|
|
VALUATION_TYPE_GROUPS = [
|
|
('all', 'All'),
|
|
('fees', 'PnL Fees'),
|
|
('goods', 'PnL Goods'),
|
|
('derivatives', 'PnL Derivatives'),
|
|
]
|
|
|
|
class ValuationBase(ModelSQL):
|
|
purchase = fields.Many2One('purchase.purchase',"Purchase")
|
|
line = fields.Many2One('purchase.line',"Purch. Line")
|
|
date = fields.Date("Date")
|
|
type = fields.Selection(VALTYPE, "Type")
|
|
reference = fields.Char("Reference")
|
|
counterparty = fields.Many2One('party.party',"Counterparty")
|
|
product = fields.Many2One('product.product',"Product")
|
|
state = fields.Char("State")
|
|
price = fields.Numeric("Price",digits=(16,4))
|
|
currency = fields.Many2One('currency.currency',"Cur")
|
|
quantity = fields.Numeric("Quantity",digits=(16,5))
|
|
unit = fields.Many2One('product.uom',"Unit")
|
|
amount = fields.Numeric("Amount",digits=(16,2))
|
|
mtm_price = fields.Numeric("Mtm Price", digits=(16,4))
|
|
mtm = fields.Numeric("Mtm",digits=(16,2))
|
|
strategy = fields.Many2One('mtm.strategy',"Strategy")
|
|
lot = fields.Many2One('lot.lot',"Lot")
|
|
base_amount = fields.Numeric("Base Amount",digits=(16,2))
|
|
rate = fields.Numeric("Rate", digits=(16,6))
|
|
|
|
@classmethod
|
|
def _get_generate_types(cls, valuation_type='all'):
|
|
type_map = {
|
|
'all': None,
|
|
'fees': {'line fee', 'pur. fee', 'sale fee', 'shipment fee'},
|
|
'goods': {
|
|
'priced', 'pur. priced', 'pur. efp',
|
|
'sale priced', 'sale efp', 'market',
|
|
},
|
|
'derivatives': {'derivative'},
|
|
}
|
|
return type_map.get(valuation_type, None)
|
|
|
|
@classmethod
|
|
def _filter_values_by_types(cls, values, selected_types):
|
|
if selected_types is None:
|
|
return values
|
|
return [value for value in values if value.get('type') in selected_types]
|
|
|
|
@classmethod
|
|
def _delete_existing(cls, line, selected_types=None):
|
|
Date = Pool().get('ir.date')
|
|
Valuation = Pool().get('valuation.valuation')
|
|
ValuationLine = Pool().get('valuation.valuation.line')
|
|
|
|
valuation_domain = [
|
|
('line', '=', line.id),
|
|
('date', '=', Date.today()),
|
|
]
|
|
valuation_line_domain = [('line', '=', line.id)]
|
|
|
|
if selected_types is not None:
|
|
valuation_domain.append(('type', 'in', list(selected_types)))
|
|
valuation_line_domain.append(('type', 'in', list(selected_types)))
|
|
|
|
valuations = Valuation.search(valuation_domain)
|
|
if valuations:
|
|
Valuation.delete(valuations)
|
|
|
|
valuation_lines = ValuationLine.search(valuation_line_domain)
|
|
if valuation_lines:
|
|
ValuationLine.delete(valuation_lines)
|
|
|
|
@classmethod
|
|
def _delete_existing_sale_line(cls, sale_line, selected_types=None):
|
|
Date = Pool().get('ir.date')
|
|
Valuation = Pool().get('valuation.valuation')
|
|
ValuationLine = Pool().get('valuation.valuation.line')
|
|
|
|
valuation_domain = [
|
|
('sale_line', '=', sale_line.id),
|
|
('date', '=', Date.today()),
|
|
]
|
|
valuation_line_domain = [('sale_line', '=', sale_line.id)]
|
|
|
|
if selected_types is not None:
|
|
valuation_domain.append(('type', 'in', list(selected_types)))
|
|
valuation_line_domain.append(('type', 'in', list(selected_types)))
|
|
|
|
valuations = Valuation.search(valuation_domain)
|
|
if valuations:
|
|
Valuation.delete(valuations)
|
|
|
|
valuation_lines = ValuationLine.search(valuation_line_domain)
|
|
if valuation_lines:
|
|
ValuationLine.delete(valuation_lines)
|
|
|
|
@classmethod
|
|
def _base_pnl(cls, *, line, lot, pnl_type, sale=None, sale_line=None):
|
|
Date = Pool().get('ir.date')
|
|
|
|
values = {
|
|
'purchase': line.purchase.id,
|
|
'line': line.id,
|
|
'type': pnl_type,
|
|
'date': Date.today(),
|
|
'lot': lot.id,
|
|
}
|
|
|
|
if sale:
|
|
values['sale'] = sale.id
|
|
if sale_line:
|
|
values['sale_line'] = sale_line.id
|
|
|
|
return values
|
|
|
|
@classmethod
|
|
def _base_sale_pnl(cls, *, sale_line, lot, pnl_type):
|
|
Date = Pool().get('ir.date')
|
|
return {
|
|
'sale': sale_line.sale.id,
|
|
'sale_line': sale_line.id,
|
|
'type': pnl_type,
|
|
'date': Date.today(),
|
|
'lot': lot.id,
|
|
}
|
|
|
|
@classmethod
|
|
def _get_strategy_mtm_price(cls, strategy, line):
|
|
total = Decimal(0)
|
|
scenario = getattr(strategy, 'scenario', None)
|
|
if not scenario:
|
|
return None
|
|
|
|
for comp in strategy.components or []:
|
|
value = Decimal(0)
|
|
|
|
if comp.price_source_type == 'curve' and comp.price_index:
|
|
value = Decimal(comp.price_index.get_price(
|
|
scenario.valuation_date,
|
|
line.unit,
|
|
strategy.currency,
|
|
last=scenario.use_last_price
|
|
))
|
|
elif comp.price_source_type == 'matrix' and comp.price_matrix:
|
|
value = Decimal(strategy._get_matrix_price(
|
|
comp, line, scenario.valuation_date))
|
|
|
|
total += value
|
|
|
|
return round(total, 4)
|
|
|
|
@staticmethod
|
|
def _supports_strategy_mtm(values):
|
|
return values and values.get('type') in {'pur. priced', 'sale priced'}
|
|
|
|
@staticmethod
|
|
def _get_basis_component_total(record):
|
|
getter = getattr(record, '_get_basis_component_price', None)
|
|
if getter:
|
|
return Decimal(getter() or 0)
|
|
return Decimal(0)
|
|
|
|
@classmethod
|
|
def _get_basis_premium_delta(cls, record):
|
|
premium_getter = getattr(record, '_get_premium_price', None)
|
|
if premium_getter:
|
|
return round(Decimal(premium_getter() or 0), 4)
|
|
total = Decimal(record.get_basis_price() or 0)
|
|
components = cls._get_basis_component_total(record)
|
|
return round(total - components, 4)
|
|
|
|
@classmethod
|
|
def _build_basis_pnl(cls, *, line, lot, sale_line, pc, sign, extra_price=Decimal(0)):
|
|
Currency = Pool().get('currency.currency')
|
|
Date = Pool().get('ir.date')
|
|
values = cls._base_pnl(
|
|
line=line,
|
|
lot=lot,
|
|
sale=sale_line.sale if sale_line else None,
|
|
sale_line=sale_line if sale_line else None,
|
|
pnl_type='sale priced' if sale_line else 'pur. priced'
|
|
)
|
|
|
|
qty = lot.get_current_quantity_converted()
|
|
|
|
price = pc.price
|
|
logger.info("TERMS:%s",line.terms)
|
|
if line.terms:
|
|
c = [t for t in line.terms if t.component == pc.price_component]
|
|
logger.info("COMPONENTS:%s",c)
|
|
if c:
|
|
price = c[0].manual_price
|
|
price = Decimal(price or 0) + Decimal(extra_price or 0)
|
|
|
|
values.update({
|
|
'reference': f"{pc.get_name()} / {pc.ratio}%",
|
|
'price': round(price, 4),
|
|
'counterparty': sale_line.sale.party.id if sale_line else line.purchase.party.id,
|
|
'product': sale_line.product.id if sale_line else line.product.id,
|
|
})
|
|
|
|
# State
|
|
if pc.unfixed_qt == 0:
|
|
values['state'] = 'fixed'
|
|
elif pc.fixed_qt == 0:
|
|
values['state'] = 'unfixed'
|
|
else:
|
|
base = sale_line.quantity_theorical if sale_line else line.quantity_theorical
|
|
values['state'] = f"part. fixed {round(pc.fixed_qt / Decimal(base) * 100, 0)}%"
|
|
|
|
if price != None:
|
|
amount = round(price * qty * Decimal(sign), 2)
|
|
base_amount = amount
|
|
currency = sale_line.sale.currency.id if sale_line else line.purchase.currency.id
|
|
rate = Decimal(1)
|
|
if line.purchase.company.currency != currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
base_amount = Currency.compute(currency,amount, line.purchase.company.currency)
|
|
rate = round(amount / (base_amount if base_amount else 1),6)
|
|
last_price = pc.get_last_price()
|
|
# mtm = round(Decimal(last_price) * qty * Decimal(sign), 2) if last_price else Decimal(0)
|
|
values.update({
|
|
'quantity': round(qty, 5),
|
|
'amount': amount,
|
|
'base_amount': base_amount,
|
|
'rate': rate,
|
|
'mtm_price': None,
|
|
'mtm': None, #round(amount - (mtm * pc.ratio / 100), 2),
|
|
'unit': sale_line.unit.id if sale_line else line.unit.id,
|
|
'currency': currency,
|
|
})
|
|
|
|
return values
|
|
|
|
@classmethod
|
|
def _build_simple_pnl(cls, *, line, lot, sale_line, price, state, sign, pnl_type):
|
|
Currency = Pool().get('currency.currency')
|
|
Date = Pool().get('ir.date')
|
|
values = cls._base_pnl(
|
|
line=line,
|
|
lot=lot,
|
|
sale=sale_line.sale if sale_line else None,
|
|
sale_line=sale_line if sale_line else None,
|
|
pnl_type=pnl_type
|
|
)
|
|
|
|
qty = lot.get_current_quantity_converted()
|
|
amount = round(price * qty * Decimal(sign), 2)
|
|
base_amount = amount
|
|
currency = sale_line.sale.currency.id if sale_line else line.purchase.currency.id
|
|
company_currency = sale_line.sale.company.currency if sale_line else line.purchase.company.currency
|
|
rate = Decimal(1)
|
|
if line.purchase.company.currency != currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
base_amount = Currency.compute(currency,amount, company_currency)
|
|
if base_amount and amount:
|
|
rate = round(amount / base_amount,6)
|
|
|
|
values.update({
|
|
'price': round(price, 4),
|
|
'quantity': round(qty, 5),
|
|
'amount': amount,
|
|
'base_amount': base_amount,
|
|
'rate': rate,
|
|
'mtm_price': None,
|
|
'mtm': Decimal(0),
|
|
'state': state,
|
|
'unit': sale_line.unit.id if sale_line else line.unit.id,
|
|
'currency': currency,
|
|
'counterparty': sale_line.sale.party.id if sale_line else line.purchase.party.id,
|
|
'product': sale_line.product.id if sale_line else line.product.id,
|
|
'reference': (
|
|
('Sale/Physic' if lot.lot_type == 'physic' else 'Sale/Open')
|
|
if sale_line else
|
|
('Purchase/Physic' if lot.lot_type == 'physic' else 'Purchase/Open')
|
|
),
|
|
})
|
|
|
|
return values
|
|
|
|
@classmethod
|
|
def create_pnl_price_from_line(cls, line):
|
|
price_lines = []
|
|
LotQt = Pool().get('lot.qt')
|
|
|
|
for lot in line.lots:
|
|
|
|
if line.price_type == 'basis':
|
|
premium_delta = cls._get_basis_premium_delta(line)
|
|
summaries = line.price_summary or []
|
|
if not summaries:
|
|
values = cls._build_simple_pnl(
|
|
line=line,
|
|
lot=lot,
|
|
sale_line=None,
|
|
price=Decimal(line.unit_price or 0) + premium_delta,
|
|
state='unfixed',
|
|
sign=-1,
|
|
pnl_type='pur. priced'
|
|
)
|
|
if values and lot.sale_line:
|
|
values['sale'] = lot.sale_line.sale.id
|
|
values['sale_line'] = lot.sale_line.id
|
|
if line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, line)
|
|
values['mtm'] = strat.get_mtm(line, values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
continue
|
|
|
|
for pc in summaries:
|
|
values = cls._build_basis_pnl(
|
|
line=line, lot=lot, sale_line=None, pc=pc, sign=-1,
|
|
extra_price=premium_delta)
|
|
if values and lot.sale_line:
|
|
values['sale'] = lot.sale_line.sale.id
|
|
values['sale_line'] = lot.sale_line.id
|
|
if line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, line)
|
|
values['mtm'] = strat.get_mtm(line,values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
elif line.price_type in ('priced', 'efp') and lot.lot_price:
|
|
values = cls._build_simple_pnl(
|
|
line=line,
|
|
lot=lot,
|
|
sale_line=None,
|
|
price=lot.lot_price,
|
|
state='fixed' if line.price_type == 'priced' else 'not fixed',
|
|
sign=-1,
|
|
pnl_type=f'pur. {line.price_type}'
|
|
)
|
|
if values and lot.sale_line:
|
|
values['sale'] = lot.sale_line.sale.id
|
|
values['sale_line'] = lot.sale_line.id
|
|
if line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, line)
|
|
values['mtm'] = strat.get_mtm(line,values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
sale_lots = [lot] if lot.sale_line else [
|
|
lqt.lot_s for lqt in LotQt.search([
|
|
('lot_p', '=', lot.id),
|
|
('lot_s', '>', 0),
|
|
('lot_quantity', '>', 0),
|
|
])
|
|
]
|
|
|
|
for sl in sale_lots:
|
|
sl_line = sl.sale_line
|
|
if not sl_line:
|
|
continue
|
|
|
|
if sl_line.price_type == 'basis':
|
|
premium_delta = cls._get_basis_premium_delta(sl_line)
|
|
for pc in sl_line.price_summary or []:
|
|
values = cls._build_basis_pnl(
|
|
line=line, lot=sl, sale_line=sl_line, pc=pc, sign=+1,
|
|
extra_price=premium_delta)
|
|
if sl_line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, sl_line)
|
|
values['mtm'] = strat.get_mtm(sl_line,values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
elif sl_line.price_type in ('priced', 'efp'):
|
|
values = cls._build_simple_pnl(
|
|
line=line,
|
|
lot=sl,
|
|
sale_line=sl_line,
|
|
price=sl.lot_price_sale,
|
|
state='fixed' if sl_line.price_type == 'priced' else 'not fixed',
|
|
sign=+1,
|
|
pnl_type=f'sale {sl_line.price_type}'
|
|
)
|
|
if sl_line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in sl_line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, sl_line)
|
|
values['mtm'] = strat.get_mtm(sl_line,values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
return price_lines
|
|
|
|
@classmethod
|
|
def _build_basis_pnl_from_sale_line(cls, *, sale_line, lot, pc, extra_price=Decimal(0)):
|
|
Currency = Pool().get('currency.currency')
|
|
Date = Pool().get('ir.date')
|
|
values = cls._base_sale_pnl(
|
|
sale_line=sale_line,
|
|
lot=lot,
|
|
pnl_type='sale priced'
|
|
)
|
|
|
|
qty = lot.get_current_quantity_converted()
|
|
price = Decimal(pc.price or 0) + Decimal(extra_price or 0)
|
|
|
|
values.update({
|
|
'reference': f"{pc.get_name()} / {pc.ratio}%",
|
|
'price': round(price, 4),
|
|
'counterparty': sale_line.sale.party.id,
|
|
'product': sale_line.product.id,
|
|
})
|
|
|
|
if pc.unfixed_qt == 0:
|
|
values['state'] = 'fixed'
|
|
elif pc.fixed_qt == 0:
|
|
values['state'] = 'unfixed'
|
|
else:
|
|
base = sale_line.quantity_theorical
|
|
values['state'] = f"part. fixed {round(pc.fixed_qt / Decimal(base) * 100, 0)}%"
|
|
|
|
if price is not None:
|
|
amount = round(price * qty, 2)
|
|
base_amount = amount
|
|
currency = sale_line.sale.currency.id
|
|
rate = Decimal(1)
|
|
if sale_line.sale.company.currency != currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
base_amount = Currency.compute(
|
|
currency, amount, sale_line.sale.company.currency)
|
|
rate = round(amount / (base_amount if base_amount else 1), 6)
|
|
values.update({
|
|
'quantity': round(qty, 5),
|
|
'amount': amount,
|
|
'base_amount': base_amount,
|
|
'rate': rate,
|
|
'mtm_price': None,
|
|
'mtm': None,
|
|
'unit': sale_line.unit.id,
|
|
'currency': currency,
|
|
})
|
|
return values
|
|
|
|
@classmethod
|
|
def _build_simple_pnl_from_sale_line(cls, *, sale_line, lot, price, state, pnl_type):
|
|
Currency = Pool().get('currency.currency')
|
|
Date = Pool().get('ir.date')
|
|
values = cls._base_sale_pnl(
|
|
sale_line=sale_line,
|
|
lot=lot,
|
|
pnl_type=pnl_type
|
|
)
|
|
|
|
qty = lot.get_current_quantity_converted()
|
|
amount = round(price * qty, 2)
|
|
base_amount = amount
|
|
currency = sale_line.sale.currency.id
|
|
company_currency = sale_line.sale.company.currency
|
|
rate = Decimal(1)
|
|
if sale_line.sale.company.currency != currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
base_amount = Currency.compute(currency, amount, company_currency)
|
|
if base_amount and amount:
|
|
rate = round(amount / base_amount, 6)
|
|
|
|
values.update({
|
|
'price': round(price, 4),
|
|
'quantity': round(qty, 5),
|
|
'amount': amount,
|
|
'base_amount': base_amount,
|
|
'rate': rate,
|
|
'mtm_price': None,
|
|
'mtm': Decimal(0),
|
|
'state': state,
|
|
'unit': sale_line.unit.id,
|
|
'currency': currency,
|
|
'counterparty': sale_line.sale.party.id,
|
|
'product': sale_line.product.id,
|
|
'reference': 'Sale/Physic' if lot.lot_type == 'physic' else 'Sale/Open',
|
|
})
|
|
|
|
return values
|
|
|
|
@classmethod
|
|
def _get_sale_lot_price(cls, sale_line, lot):
|
|
if lot.lot_price_sale is not None:
|
|
return lot.lot_price_sale
|
|
return sale_line.unit_price
|
|
|
|
@classmethod
|
|
def create_pnl_price_from_sale_line(cls, sale_line):
|
|
price_lines = []
|
|
|
|
for lot in sale_line.lots or []:
|
|
if sale_line.price_type == 'basis':
|
|
summaries = sale_line.price_summary or []
|
|
premium_delta = cls._get_basis_premium_delta(sale_line)
|
|
if not summaries:
|
|
values = cls._build_simple_pnl_from_sale_line(
|
|
sale_line=sale_line,
|
|
lot=lot,
|
|
price=Decimal(sale_line.unit_price or 0) + premium_delta,
|
|
state='unfixed',
|
|
pnl_type='sale priced'
|
|
)
|
|
if sale_line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in sale_line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, sale_line)
|
|
values['mtm'] = strat.get_mtm(sale_line, values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
continue
|
|
|
|
for pc in summaries:
|
|
values = cls._build_basis_pnl_from_sale_line(
|
|
sale_line=sale_line, lot=lot, pc=pc,
|
|
extra_price=premium_delta)
|
|
if sale_line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in sale_line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, sale_line)
|
|
values['mtm'] = strat.get_mtm(sale_line, values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
elif sale_line.price_type in ('priced', 'efp'):
|
|
price = cls._get_sale_lot_price(sale_line, lot)
|
|
if price is None:
|
|
continue
|
|
values = cls._build_simple_pnl_from_sale_line(
|
|
sale_line=sale_line,
|
|
lot=lot,
|
|
price=price,
|
|
state='fixed' if sale_line.price_type == 'priced' else 'not fixed',
|
|
pnl_type=f'sale {sale_line.price_type}'
|
|
)
|
|
if sale_line.mtm and cls._supports_strategy_mtm(values):
|
|
for strat in sale_line.mtm:
|
|
values['mtm_price'] = cls._get_strategy_mtm_price(strat, sale_line)
|
|
values['mtm'] = strat.get_mtm(sale_line, values['quantity'])
|
|
values['strategy'] = strat
|
|
|
|
if values:
|
|
price_lines.append(values)
|
|
else:
|
|
if values:
|
|
price_lines.append(values)
|
|
|
|
return price_lines
|
|
|
|
@classmethod
|
|
def group_fees_by_type_supplier(cls,line,fees):
|
|
grouped = defaultdict(list)
|
|
|
|
# Regrouper par (type, supplier)
|
|
for fee in fees:
|
|
key = (fee.product, fee.supplier)
|
|
grouped[key].append(fee)
|
|
result = []
|
|
for key, fee_list in grouped.items():
|
|
ordered_fees = [f for f in fee_list if f.type == 'ordered']
|
|
if ordered_fees:
|
|
result.extend(ordered_fees)
|
|
else:
|
|
budgeted_fees = [f for f in fee_list if f.type == 'budgeted']
|
|
result.extend(budgeted_fees)
|
|
return result
|
|
|
|
@classmethod
|
|
def create_pnl_fee_from_line(cls, line):
|
|
fee_lines = []
|
|
Date = Pool().get('ir.date')
|
|
Currency = Pool().get('currency.currency')
|
|
FeeLots = Pool().get('fee.lots')
|
|
#if line is matched with sale_line we should add the open sale side
|
|
sale_lines = line.get_matched_lines() or []
|
|
sale_open_lots = tuple(s.lot_s for s in sale_lines if s.lot_s)
|
|
all_lots = (line.lots or ()) + sale_open_lots
|
|
for lot in all_lots:
|
|
fl = FeeLots.search([('lot', '=', lot.id)])
|
|
if not fl:
|
|
continue
|
|
|
|
fees = [e.fee for e in fl]
|
|
for sf in cls.group_fees_by_type_supplier(line, fees):
|
|
sign = -1 if sf.p_r == 'pay' else 1
|
|
qty = round(lot.get_current_quantity_converted(), 5)
|
|
if sf.mode == 'ppack' or sf.mode == 'rate':
|
|
price = sf.price
|
|
amount = sf.amount * sign
|
|
elif sf.mode == 'lumpsum':
|
|
price = sf.price
|
|
amount = sf.price * sign
|
|
qty = 1
|
|
else:
|
|
price = Decimal(sf.get_price_per_qt())
|
|
amount = round(price * lot.get_current_quantity_converted() * sign, 2)
|
|
if sf.currency != line.purchase.currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
price = Currency.compute(sf.currency, price, line.purchase.currency)
|
|
fee_lines.append({
|
|
'lot': lot.id,
|
|
'sale': lot.sale_line.sale.id if lot.sale_line else None,
|
|
'purchase': line.purchase.id,
|
|
'line': line.id,
|
|
'type': (
|
|
'shipment fee' if sf.shipment_in
|
|
else 'sale fee' if sf.sale_line
|
|
else 'pur. fee'
|
|
),
|
|
'date': Date.today(),
|
|
'price': price,
|
|
'counterparty': sf.supplier.id,
|
|
'reference': f"{sf.product.name}/{'Physic' if lot.lot_type == 'physic' else 'Open'}",
|
|
'product': sf.product.id,
|
|
'state': sf.type,
|
|
'quantity': qty,
|
|
'amount': amount,
|
|
'mtm_price': None,
|
|
'mtm': None,
|
|
'strategy': None,
|
|
'unit': sf.unit.id if sf.unit else line.unit.id,
|
|
'currency': sf.currency.id,
|
|
})
|
|
|
|
return fee_lines
|
|
|
|
@classmethod
|
|
def create_pnl_fee_from_sale_line(cls, sale_line):
|
|
fee_lines = []
|
|
Date = Pool().get('ir.date')
|
|
Currency = Pool().get('currency.currency')
|
|
FeeLots = Pool().get('fee.lots')
|
|
|
|
for lot in sale_line.lots or ():
|
|
fl = FeeLots.search([('lot', '=', lot.id)])
|
|
if not fl:
|
|
continue
|
|
|
|
fees = [
|
|
e.fee for e in fl
|
|
if e.fee and (not e.fee.sale_line or e.fee.sale_line.id == sale_line.id)
|
|
]
|
|
for sf in cls.group_fees_by_type_supplier(sale_line, fees):
|
|
sign = -1 if sf.p_r == 'pay' else 1
|
|
qty = round(lot.get_current_quantity_converted(), 5)
|
|
if sf.mode == 'ppack' or sf.mode == 'rate':
|
|
price = sf.price
|
|
amount = sf.amount * sign
|
|
elif sf.mode == 'lumpsum':
|
|
price = sf.price
|
|
amount = sf.price * sign
|
|
qty = 1
|
|
else:
|
|
price = Decimal(sf.get_price_per_qt())
|
|
amount = round(price * lot.get_current_quantity_converted() * sign, 2)
|
|
if sf.currency != sale_line.sale.currency:
|
|
with Transaction().set_context(date=Date.today()):
|
|
price = Currency.compute(sf.currency, price, sale_line.sale.currency)
|
|
fee_lines.append({
|
|
'lot': lot.id,
|
|
'sale': sale_line.sale.id,
|
|
'sale_line': sale_line.id,
|
|
'type': (
|
|
'shipment fee' if sf.shipment_in
|
|
else 'sale fee'
|
|
),
|
|
'date': Date.today(),
|
|
'price': price,
|
|
'counterparty': sf.supplier.id,
|
|
'reference': f"{sf.product.name}/{'Physic' if lot.lot_type == 'physic' else 'Open'}",
|
|
'product': sf.product.id,
|
|
'state': sf.type,
|
|
'quantity': qty,
|
|
'amount': amount,
|
|
'mtm_price': None,
|
|
'mtm': None,
|
|
'strategy': None,
|
|
'unit': sf.unit.id if sf.unit else sale_line.unit.id,
|
|
'currency': sf.currency.id,
|
|
})
|
|
|
|
return fee_lines
|
|
|
|
@classmethod
|
|
def create_pnl_der_from_line(cls, line):
|
|
Date = Pool().get('ir.date')
|
|
der_lines = []
|
|
|
|
for d in line.derivatives or []:
|
|
price = Decimal(d.price_index.get_price_per_qt(
|
|
d.price, line.unit, line.purchase.currency
|
|
))
|
|
|
|
mtm_price = Decimal(d.price_index.get_price(
|
|
Date.today(), line.unit, line.purchase.currency, True
|
|
))
|
|
|
|
der_lines.append({
|
|
'purchase': line.purchase.id,
|
|
'line': line.id,
|
|
'type': 'derivative',
|
|
'date': Date.today(),
|
|
'reference': d.price_index.price_index,
|
|
'price': round(price, 4),
|
|
'counterparty': d.party.id,
|
|
'product': d.product.id,
|
|
'state': 'fixed',
|
|
'quantity': round(d.quantity, 5),
|
|
'amount': round(price * d.quantity * Decimal(-1), 2),
|
|
'mtm_price': round(mtm_price, 4),
|
|
'mtm': round((price * d.quantity * Decimal(-1)) - (mtm_price * d.quantity * Decimal(-1)), 2),
|
|
'unit': line.unit.id,
|
|
'currency': line.purchase.currency.id,
|
|
})
|
|
|
|
return der_lines
|
|
|
|
@classmethod
|
|
def create_pnl_der_from_sale_line(cls, sale_line):
|
|
Date = Pool().get('ir.date')
|
|
der_lines = []
|
|
|
|
for d in sale_line.derivatives or []:
|
|
price = Decimal(d.price_index.get_price_per_qt(
|
|
d.price, sale_line.unit, sale_line.sale.currency
|
|
))
|
|
|
|
mtm_price = Decimal(d.price_index.get_price(
|
|
Date.today(), sale_line.unit, sale_line.sale.currency, True
|
|
))
|
|
|
|
der_lines.append({
|
|
'sale': sale_line.sale.id,
|
|
'sale_line': sale_line.id,
|
|
'type': 'derivative',
|
|
'date': Date.today(),
|
|
'reference': d.price_index.price_index,
|
|
'price': round(price, 4),
|
|
'counterparty': d.party.id,
|
|
'product': d.product.id,
|
|
'state': 'fixed',
|
|
'quantity': round(d.quantity, 5),
|
|
'amount': round(price * d.quantity * Decimal(-1), 2),
|
|
'mtm_price': round(mtm_price, 4),
|
|
'mtm': round((price * d.quantity * Decimal(-1)) - (mtm_price * d.quantity * Decimal(-1)), 2),
|
|
'unit': sale_line.unit.id,
|
|
'currency': sale_line.sale.currency.id,
|
|
})
|
|
|
|
return der_lines
|
|
|
|
@classmethod
|
|
def generate(cls, line, valuation_type='all'):
|
|
selected_types = cls._get_generate_types(valuation_type)
|
|
cls._delete_existing(line, selected_types=selected_types)
|
|
values = []
|
|
values.extend(cls.create_pnl_fee_from_line(line))
|
|
values.extend(cls.create_pnl_price_from_line(line))
|
|
values.extend(cls.create_pnl_der_from_line(line))
|
|
values = cls._filter_values_by_types(values, selected_types)
|
|
|
|
if values:
|
|
Valuation = Pool().get('valuation.valuation')
|
|
ValuationLine = Pool().get('valuation.valuation.line')
|
|
Valuation.create(values)
|
|
ValuationLine.create(values)
|
|
|
|
@classmethod
|
|
def generate_from_sale_line(cls, sale_line, valuation_type='all'):
|
|
selected_types = cls._get_generate_types(valuation_type)
|
|
cls._delete_existing_sale_line(sale_line, selected_types=selected_types)
|
|
values = []
|
|
values.extend(cls.create_pnl_fee_from_sale_line(sale_line))
|
|
values.extend(cls.create_pnl_price_from_sale_line(sale_line))
|
|
values.extend(cls.create_pnl_der_from_sale_line(sale_line))
|
|
values = cls._filter_values_by_types(values, selected_types)
|
|
|
|
if values:
|
|
Valuation = Pool().get('valuation.valuation')
|
|
ValuationLine = Pool().get('valuation.valuation.line')
|
|
Valuation.create(values)
|
|
ValuationLine.create(values)
|
|
|
|
class Valuation(ValuationBase, ModelView):
|
|
"Valuation"
|
|
__name__ = 'valuation.valuation'
|
|
|
|
@classmethod
|
|
def get_totals(cls):
|
|
cursor = Transaction().connection.cursor()
|
|
table = cls.__table__()
|
|
|
|
sql = f"""
|
|
WITH totals AS (
|
|
SELECT
|
|
date,
|
|
SUM(amount) AS total_amount
|
|
FROM {table}
|
|
WHERE line IS NOT NULL
|
|
OR sale_line IS NOT NULL
|
|
GROUP BY date
|
|
),
|
|
ranked AS (
|
|
SELECT
|
|
date,
|
|
total_amount,
|
|
LAG(total_amount) OVER (ORDER BY date) AS previous_total,
|
|
ROW_NUMBER() OVER (ORDER BY date DESC) AS rn
|
|
FROM totals
|
|
)
|
|
SELECT
|
|
total_amount AS last_total,
|
|
total_amount - previous_total AS last_variation
|
|
FROM ranked
|
|
WHERE rn = 1;
|
|
"""
|
|
|
|
cursor.execute(sql)
|
|
row = cursor.fetchone()
|
|
if not row:
|
|
return Decimal(0), Decimal(0)
|
|
|
|
last_total, last_variation = row
|
|
|
|
return last_total or Decimal(0), last_variation or Decimal(0)
|
|
|
|
class ValuationLine(ValuationBase, ModelView):
|
|
"Last Valuation"
|
|
__name__ = 'valuation.valuation.line'
|
|
|
|
class ValuationDyn(ModelSQL,ModelView):
|
|
"Valuation"
|
|
__name__ = 'valuation.valuation.dyn'
|
|
|
|
r_purchase = fields.Many2One('purchase.purchase',"Purchase")
|
|
r_line = fields.Many2One('purchase.line',"Line")
|
|
r_date = fields.Date("Date")
|
|
r_type = fields.Selection(VALTYPE, "Type")
|
|
r_reference = fields.Char("Reference")
|
|
r_counterparty = fields.Many2One('party.party',"Counterparty")
|
|
r_product = fields.Many2One('product.product',"Product")
|
|
r_state = fields.Char("State")
|
|
r_price = fields.Numeric("Price",digits='r_unit')
|
|
r_currency = fields.Many2One('currency.currency',"Cur")
|
|
r_quantity = fields.Numeric("Quantity",digits='r_unit')
|
|
r_unit = fields.Many2One('product.uom',"Unit")
|
|
r_amount = fields.Numeric("Amount",digits='r_unit')
|
|
r_base_amount = fields.Numeric("Base Amount",digits='r_unit')
|
|
r_rate = fields.Numeric("Rate",digits=(16,6))
|
|
r_mtm_price = fields.Numeric("Mtm Price",digits='r_unit')
|
|
r_mtm = fields.Numeric("Mtm",digits='r_unit')
|
|
r_strategy = fields.Many2One('mtm.strategy',"Strategy")
|
|
r_lot = fields.Many2One('lot.lot',"Lot")
|
|
|
|
@classmethod
|
|
def table_query(cls):
|
|
Valuation = Pool().get('valuation.valuation.line')
|
|
val = Valuation.__table__()
|
|
context = Transaction().context
|
|
group_pnl = context.get('group_pnl')
|
|
wh = (val.id > 0)
|
|
|
|
query = val.select(
|
|
Literal(0).as_('create_uid'),
|
|
CurrentTimestamp().as_('create_date'),
|
|
Literal(None).as_('write_uid'),
|
|
Literal(None).as_('write_date'),
|
|
Max(val.id).as_('id'),
|
|
Max(val.purchase).as_('r_purchase'),
|
|
Max(val.line).as_('r_line'),
|
|
Max(val.date).as_('r_date'),
|
|
val.type.as_('r_type'),
|
|
Max(val.reference).as_('r_reference'),
|
|
val.counterparty.as_('r_counterparty'),
|
|
Max(val.product).as_('r_product'),
|
|
val.state.as_('r_state'),
|
|
Avg(val.price).as_('r_price'),
|
|
Max(val.currency).as_('r_currency'),
|
|
Sum(val.quantity).as_('r_quantity'),
|
|
Max(val.unit).as_('r_unit'),
|
|
Sum(val.amount).as_('r_amount'),
|
|
Sum(val.base_amount).as_('r_base_amount'),
|
|
Sum(val.rate).as_('r_rate'),
|
|
Avg(val.mtm_price).as_('r_mtm_price'),
|
|
Sum(val.mtm).as_('r_mtm'),
|
|
Max(val.strategy).as_('r_strategy'),
|
|
Max(val.lot).as_('r_lot'),
|
|
where=wh,
|
|
group_by=[val.type,val.counterparty,val.state])
|
|
|
|
return query
|
|
|
|
class ValuationReport(ValuationBase, ModelView):
|
|
"Valuation Report"
|
|
__name__ = 'valuation.report'
|
|
|
|
@classmethod
|
|
def table_query(cls):
|
|
Valuation = Pool().get('valuation.valuation')
|
|
Date = Pool().get('ir.date')
|
|
val = Valuation.__table__()
|
|
context = Transaction().context
|
|
valuation_date = context.get('valuation_date')
|
|
strategy = context.get('strategy')
|
|
if not valuation_date:
|
|
valuation_date = Date.today()
|
|
wh = (val.date == valuation_date)
|
|
if strategy:
|
|
wh &= (val.strategy == strategy)
|
|
|
|
query = val.select(
|
|
Literal(0).as_('create_uid'),
|
|
CurrentTimestamp().as_('create_date'),
|
|
Literal(None).as_('write_uid'),
|
|
Literal(None).as_('write_date'),
|
|
val.id.as_('id'),
|
|
val.purchase.as_('purchase'),
|
|
val.sale.as_('sale'),
|
|
val.sale_line.as_('sale_line'),
|
|
val.line.as_('line'),
|
|
val.date.as_('date'),
|
|
val.type.as_('type'),
|
|
val.reference.as_('reference'),
|
|
val.counterparty.as_('counterparty'),
|
|
val.product.as_('product'),
|
|
val.state.as_('state'),
|
|
val.price.as_('price'),
|
|
val.currency.as_('currency'),
|
|
val.quantity.as_('quantity'),
|
|
val.unit.as_('unit'),
|
|
val.amount.as_('amount'),
|
|
val.base_amount.as_('base_amount'),
|
|
val.rate.as_('rate'),
|
|
val.mtm_price.as_('mtm_price'),
|
|
val.mtm.as_('mtm'),
|
|
val.strategy.as_('strategy'),
|
|
val.lot.as_('lot'),
|
|
where=wh)
|
|
|
|
return query
|
|
|
|
class ValuationReportContext(ModelView):
|
|
"Valuation Report Context"
|
|
__name__ = 'valuation.report.context'
|
|
|
|
valuation_date = fields.Date("Valuation date")
|
|
supplier = fields.Many2One('party.party',"Supplier")
|
|
client = fields.Many2One('party.party',"Client")
|
|
product = fields.Many2One('product.product',"Product")
|
|
purchase = fields.Many2One('purchase.purchase', "Purchase")
|
|
sale = fields.Many2One('sale.sale',"Sale")
|
|
state = fields.Selection([
|
|
('all', 'All'),
|
|
('open', 'Open'),
|
|
('fixed', 'Fixed'),
|
|
('hedged', 'Hedged')
|
|
], 'State')
|
|
strategy = fields.Many2One('mtm.strategy',"Strategy")
|
|
|
|
@classmethod
|
|
def default_valuation_date(cls):
|
|
pool = Pool()
|
|
Date = pool.get('ir.date')
|
|
return Date.today()
|
|
|
|
@classmethod
|
|
def default_state(cls):
|
|
return 'all'
|
|
|
|
|
|
class ValuationProcessDimension(ModelView):
|
|
"Valuation Process Dimension"
|
|
__name__ = 'valuation.process.dimension'
|
|
|
|
start = fields.Many2One('valuation.process.start', "Start")
|
|
dimension = fields.Many2One(
|
|
'analytic.dimension',
|
|
'Dimension',
|
|
required=True
|
|
)
|
|
value = fields.Many2One(
|
|
'analytic.dimension.value',
|
|
'Value',
|
|
required=True,
|
|
domain=[
|
|
('dimension', '=', Eval('dimension')),
|
|
],
|
|
depends=['dimension']
|
|
)
|
|
|
|
|
|
class ValuationProcessStart(ModelView):
|
|
"Valuation Process Start"
|
|
__name__ = 'valuation.process.start'
|
|
|
|
purchase_from_date = fields.Date("Purchase From Date")
|
|
purchase_to_date = fields.Date("Purchase To Date")
|
|
sale_from_date = fields.Date("Sale From Date")
|
|
sale_to_date = fields.Date("Sale To Date")
|
|
purchase_numbers = fields.Char("Purchase Numbers")
|
|
sale_numbers = fields.Char("Sale Numbers")
|
|
analytic_dimensions = fields.One2Many(
|
|
'valuation.process.dimension',
|
|
'start',
|
|
'Analytic Dimensions'
|
|
)
|
|
valuation_type = fields.Selection(
|
|
VALUATION_TYPE_GROUPS,
|
|
"Type",
|
|
required=True
|
|
)
|
|
|
|
@classmethod
|
|
def default_valuation_type(cls):
|
|
return 'all'
|
|
|
|
|
|
class ValuationProcessResult(ModelView):
|
|
"Valuation Process Result"
|
|
__name__ = 'valuation.process.result'
|
|
|
|
message = fields.Text("Message", readonly=True)
|
|
|
|
|
|
class ValuationProcess(Wizard):
|
|
"Process Valuation"
|
|
__name__ = 'valuation.process'
|
|
|
|
start = StateView(
|
|
'valuation.process.start',
|
|
'purchase_trade.valuation_process_start_view_form',
|
|
[
|
|
Button('Cancel', 'end', 'tryton-cancel'),
|
|
Button('Process', 'process', 'tryton-ok', default=True),
|
|
]
|
|
)
|
|
process = StateTransition()
|
|
result = StateView(
|
|
'valuation.process.result',
|
|
'purchase_trade.valuation_process_result_view_form',
|
|
[
|
|
Button('Close', 'end', 'tryton-cancel'),
|
|
Button('See Valuation', 'open_report', 'tryton-go-next', default=True),
|
|
]
|
|
)
|
|
open_report = StateAction('purchase_trade.act_valuation_form')
|
|
_result_message = None
|
|
|
|
@staticmethod
|
|
def _parse_numbers(text):
|
|
if not text:
|
|
return []
|
|
return [item for item in re.split(r'[\s,;]+', text) if item]
|
|
|
|
@staticmethod
|
|
def _matches_dimensions(record, dimension_filters):
|
|
assignments = getattr(record, 'analytic_dimensions', []) or []
|
|
assignment_pairs = {
|
|
(assignment.dimension.id, assignment.value.id)
|
|
for assignment in assignments
|
|
if assignment.dimension and assignment.value
|
|
}
|
|
return all(
|
|
(dimension.id, value.id) in assignment_pairs
|
|
for dimension, value in dimension_filters
|
|
)
|
|
|
|
@classmethod
|
|
def _get_dimension_filters(cls, start):
|
|
return [
|
|
(line.dimension, line.value)
|
|
for line in start.analytic_dimensions or []
|
|
if line.dimension and line.value
|
|
]
|
|
|
|
@classmethod
|
|
def _search_purchase_ids(cls, start, dimension_filters):
|
|
Purchase = Pool().get('purchase.purchase')
|
|
domain = []
|
|
numbers = cls._parse_numbers(start.purchase_numbers)
|
|
if start.purchase_from_date:
|
|
domain.append(('purchase_date', '>=', start.purchase_from_date))
|
|
if start.purchase_to_date:
|
|
domain.append(('purchase_date', '<=', start.purchase_to_date))
|
|
if numbers:
|
|
domain.append(('number', 'in', numbers))
|
|
|
|
purchases = Purchase.search(domain)
|
|
if dimension_filters:
|
|
purchases = [
|
|
purchase for purchase in purchases
|
|
if cls._matches_dimensions(purchase, dimension_filters)
|
|
]
|
|
return {purchase.id for purchase in purchases}
|
|
|
|
@classmethod
|
|
def _search_sale_ids(cls, start, dimension_filters):
|
|
Sale = Pool().get('sale.sale')
|
|
domain = []
|
|
numbers = cls._parse_numbers(start.sale_numbers)
|
|
if start.sale_from_date:
|
|
domain.append(('sale_date', '>=', start.sale_from_date))
|
|
if start.sale_to_date:
|
|
domain.append(('sale_date', '<=', start.sale_to_date))
|
|
if numbers:
|
|
domain.append(('number', 'in', numbers))
|
|
|
|
sales = Sale.search(domain)
|
|
if dimension_filters:
|
|
sales = [sale for sale in sales if cls._matches_dimensions(sale, dimension_filters)]
|
|
return {sale.id for sale in sales}
|
|
|
|
@classmethod
|
|
def _purchase_line_ids_from_purchase_ids(cls, purchase_ids):
|
|
if not purchase_ids:
|
|
return set()
|
|
PurchaseLine = Pool().get('purchase.line')
|
|
return {
|
|
line.id for line in PurchaseLine.search([('purchase', 'in', list(purchase_ids))])
|
|
}
|
|
|
|
@classmethod
|
|
def _purchase_line_ids_from_sale_ids(cls, sale_ids):
|
|
if not sale_ids:
|
|
return set()
|
|
SaleLine = Pool().get('sale.line')
|
|
purchase_line_ids = set()
|
|
sale_lines = SaleLine.search([('sale', 'in', list(sale_ids))])
|
|
for sale_line in sale_lines:
|
|
for matched_line in sale_line.get_matched_lines() or []:
|
|
if matched_line.lot_p and matched_line.lot_p.line:
|
|
purchase_line_ids.add(matched_line.lot_p.line.id)
|
|
return purchase_line_ids
|
|
|
|
@classmethod
|
|
def _sale_line_is_unmatched(cls, sale_line):
|
|
for matched_line in sale_line.get_matched_lines() or []:
|
|
if matched_line.lot_p and matched_line.lot_p.line:
|
|
return False
|
|
return True
|
|
|
|
@classmethod
|
|
def _sale_line_ids_from_sale_ids(cls, sale_ids, unmatched_only=False):
|
|
if not sale_ids:
|
|
return set()
|
|
SaleLine = Pool().get('sale.line')
|
|
sale_lines = SaleLine.search([('sale', 'in', list(sale_ids))])
|
|
if unmatched_only:
|
|
sale_lines = [
|
|
sale_line for sale_line in sale_lines
|
|
if cls._sale_line_is_unmatched(sale_line)
|
|
]
|
|
return {sale_line.id for sale_line in sale_lines}
|
|
|
|
@classmethod
|
|
def _get_target_sale_line_ids(cls, start):
|
|
Sale = Pool().get('sale.sale')
|
|
dimension_filters = cls._get_dimension_filters(start)
|
|
has_purchase_filters = bool(
|
|
start.purchase_from_date
|
|
or start.purchase_to_date
|
|
or cls._parse_numbers(start.purchase_numbers)
|
|
)
|
|
has_sale_filters = bool(
|
|
start.sale_from_date
|
|
or start.sale_to_date
|
|
or cls._parse_numbers(start.sale_numbers)
|
|
)
|
|
|
|
if has_sale_filters:
|
|
sale_ids = cls._search_sale_ids(start, dimension_filters)
|
|
return cls._sale_line_ids_from_sale_ids(sale_ids, unmatched_only=True)
|
|
if dimension_filters and not has_purchase_filters:
|
|
sale_ids = cls._search_sale_ids(start, dimension_filters)
|
|
return cls._sale_line_ids_from_sale_ids(sale_ids, unmatched_only=True)
|
|
if not has_purchase_filters and not has_sale_filters and not dimension_filters:
|
|
sale_ids = {sale.id for sale in Sale.search([])}
|
|
return cls._sale_line_ids_from_sale_ids(sale_ids, unmatched_only=True)
|
|
return set()
|
|
|
|
@classmethod
|
|
def _get_target_purchase_line_ids(cls, start):
|
|
PurchaseLine = Pool().get('purchase.line')
|
|
dimension_filters = cls._get_dimension_filters(start)
|
|
has_purchase_filters = bool(
|
|
start.purchase_from_date
|
|
or start.purchase_to_date
|
|
or cls._parse_numbers(start.purchase_numbers)
|
|
)
|
|
has_sale_filters = bool(
|
|
start.sale_from_date
|
|
or start.sale_to_date
|
|
or cls._parse_numbers(start.sale_numbers)
|
|
)
|
|
|
|
purchase_side_ids = cls._purchase_line_ids_from_purchase_ids(
|
|
cls._search_purchase_ids(
|
|
start,
|
|
dimension_filters if (dimension_filters and (has_purchase_filters or not has_sale_filters)) else [],
|
|
)
|
|
) if (has_purchase_filters or (dimension_filters and not has_sale_filters)) else set()
|
|
|
|
sale_side_ids = cls._purchase_line_ids_from_sale_ids(
|
|
cls._search_sale_ids(
|
|
start,
|
|
dimension_filters if (dimension_filters and (has_sale_filters or not has_purchase_filters)) else [],
|
|
)
|
|
) if (has_sale_filters or (dimension_filters and not has_purchase_filters)) else set()
|
|
|
|
if has_purchase_filters and has_sale_filters:
|
|
target_ids = purchase_side_ids & sale_side_ids
|
|
elif has_purchase_filters:
|
|
target_ids = purchase_side_ids
|
|
elif has_sale_filters:
|
|
target_ids = sale_side_ids
|
|
elif dimension_filters:
|
|
target_ids = purchase_side_ids | sale_side_ids
|
|
else:
|
|
target_ids = {line.id for line in PurchaseLine.search([])}
|
|
|
|
return target_ids
|
|
|
|
def transition_process(self):
|
|
PurchaseLine = Pool().get('purchase.line')
|
|
SaleLine = Pool().get('sale.line')
|
|
target_ids = self._get_target_purchase_line_ids(self.start)
|
|
target_sale_line_ids = self._get_target_sale_line_ids(self.start)
|
|
lines = PurchaseLine.browse(list(target_ids))
|
|
sale_lines = SaleLine.browse(list(target_sale_line_ids))
|
|
purchase_ids = {line.purchase.id for line in lines if line.purchase}
|
|
sale_ids = set()
|
|
for line in lines:
|
|
for matched_line in line.get_matched_lines() or []:
|
|
if matched_line.lot_s and matched_line.lot_s.sale_line:
|
|
sale_ids.add(matched_line.lot_s.sale_line.sale.id)
|
|
|
|
Valuation.generate(line, valuation_type=self.start.valuation_type)
|
|
|
|
for sale_line in sale_lines:
|
|
sale_ids.add(sale_line.sale.id)
|
|
Valuation.generate_from_sale_line(
|
|
sale_line, valuation_type=self.start.valuation_type)
|
|
|
|
self._result_message = (
|
|
f"Processed {len(lines)} purchase line(s) "
|
|
f"and {len(sale_lines)} unmatched sale line(s) "
|
|
f"from {len(purchase_ids)} purchase(s) "
|
|
f"and {len(sale_ids)} sale(s)."
|
|
)
|
|
return 'result'
|
|
|
|
def default_result(self, fields):
|
|
return {
|
|
'message': self._result_message or 'No valuation was processed.',
|
|
}
|
|
|
|
def do_open_report(self, action):
|
|
Date = Pool().get('ir.date')
|
|
action['pyson_context'] = PYSONEncoder().encode({
|
|
'valuation_date': Date.today(),
|
|
})
|
|
return action, {}
|