1944 lines
72 KiB
Python
1944 lines
72 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))
|
|
amount_prev = fields.Numeric("Amount -1", digits=(16, 2))
|
|
mtm_price = fields.Numeric("Mtm Price", digits=(16,4))
|
|
mtm_price_prev = fields.Numeric("Mtm Price -1", digits=(16,4))
|
|
mtm_curve = fields.Many2One('price.price', "Curve")
|
|
mtm = fields.Numeric("Mtm",digits=(16,2))
|
|
strategy = fields.Many2One('mtm.strategy',"Strategy")
|
|
lot = fields.Many2One('lot.lot',"Lot")
|
|
shipment_in = fields.Many2One('stock.shipment.in', "Shipment In")
|
|
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 _is_finished(cls, record):
|
|
return bool(getattr(record, 'finished', False))
|
|
|
|
@classmethod
|
|
def _valuation_lots(cls, record):
|
|
lots = tuple(record.lots or ())
|
|
if cls._is_finished(record):
|
|
return tuple(lot for lot in lots if lot.lot_type == 'physic')
|
|
return lots
|
|
|
|
@classmethod
|
|
def _ignore_finished_open_lot(cls, line, lot):
|
|
return cls._is_finished(line) and lot.lot_type != 'physic'
|
|
|
|
@classmethod
|
|
def _lot_quantity(cls, lot):
|
|
return Decimal(str(lot.get_current_quantity_converted() or 0))
|
|
|
|
@classmethod
|
|
def _lot_shipment_in_id(cls, lot):
|
|
shipment = getattr(lot, 'lot_shipment_in', None)
|
|
shipment_id = cls._record_id(shipment) if shipment else None
|
|
return shipment_id if isinstance(shipment_id, int) else None
|
|
|
|
@classmethod
|
|
def _lotqt_shipment_in_id(cls, lqt):
|
|
shipment = getattr(lqt, 'lot_shipment_in', None)
|
|
shipment_id = cls._record_id(shipment) if shipment else None
|
|
return shipment_id if isinstance(shipment_id, int) else None
|
|
|
|
@classmethod
|
|
def _lotqt_quantity(cls, lqt, line):
|
|
quantity = abs(Decimal(str(getattr(lqt, 'lot_quantity', 0) or 0)))
|
|
lot_unit = getattr(lqt, 'lot_unit', None)
|
|
line_unit = getattr(line, 'unit', None)
|
|
if not lot_unit or not line_unit:
|
|
return quantity
|
|
lot_unit_id = cls._record_id(lot_unit)
|
|
line_unit_id = cls._record_id(line_unit)
|
|
if lot_unit_id == line_unit_id:
|
|
return quantity
|
|
Uom = Pool().get('product.uom')
|
|
factor = None
|
|
rate = None
|
|
lot_category = getattr(lot_unit, 'category', None)
|
|
line_category = getattr(line_unit, 'category', None)
|
|
if (lot_category and line_category
|
|
and cls._record_id(lot_category) != cls._record_id(line_category)):
|
|
factor = 1
|
|
rate = 1
|
|
return Decimal(str(Uom.compute_qty(
|
|
lot_unit, float(quantity), line_unit, True, factor, rate)))
|
|
|
|
@classmethod
|
|
def _purchase_lot_segments(cls, line, lot, LotQt):
|
|
if getattr(lot, 'lot_type', None) == 'physic':
|
|
sale_lot = lot if getattr(lot, 'sale_line', None) else None
|
|
return [{
|
|
'quantity': None,
|
|
'sale_quantity': None,
|
|
'sale_lot': sale_lot,
|
|
'sale_line': getattr(sale_lot, 'sale_line', None),
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
lot_id = getattr(lot, 'id', None)
|
|
if not lot_id:
|
|
return [{
|
|
'quantity': None,
|
|
'sale_quantity': None,
|
|
'sale_lot': None,
|
|
'sale_line': None,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
lqts = LotQt.search([
|
|
('lot_p', '=', lot_id),
|
|
('lot_quantity', '>', 0),
|
|
])
|
|
if not lqts:
|
|
matched_sale_line = cls._get_matched_sale_line_from_purchase_lot(
|
|
lot, LotQt=LotQt)
|
|
return [{
|
|
'quantity': None,
|
|
'sale_quantity': None,
|
|
'sale_lot': None,
|
|
'sale_line': matched_sale_line,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
segments = []
|
|
for lqt in lqts:
|
|
sale_lot = getattr(lqt, 'lot_s', None)
|
|
sale_line = getattr(sale_lot, 'sale_line', None)
|
|
segments.append({
|
|
'quantity': cls._lotqt_quantity(lqt, line),
|
|
'sale_quantity': (
|
|
cls._lotqt_quantity(lqt, sale_line)
|
|
if sale_line else None),
|
|
'sale_lot': sale_lot,
|
|
'sale_line': sale_line,
|
|
'shipment_in': cls._lotqt_shipment_in_id(lqt),
|
|
})
|
|
return segments
|
|
|
|
@classmethod
|
|
def _ignore_empty_open_fee_lot(cls, lot):
|
|
return lot.lot_type != 'physic' and cls._lot_quantity(lot) == 0
|
|
|
|
@classmethod
|
|
def _get_matched_sale_line_from_purchase_lot(cls, lot, LotQt=None):
|
|
sale_line = getattr(lot, 'sale_line', None)
|
|
if sale_line:
|
|
return sale_line
|
|
|
|
lot_id = getattr(lot, 'id', None)
|
|
if not lot_id:
|
|
return None
|
|
|
|
if LotQt is None:
|
|
LotQt = Pool().get('lot.qt')
|
|
|
|
sale_lines = []
|
|
seen = set()
|
|
for lqt in LotQt.search([
|
|
('lot_p', '=', lot_id),
|
|
('lot_s', '>', 0),
|
|
('lot_quantity', '>', 0),
|
|
]):
|
|
lot_s = getattr(lqt, 'lot_s', None)
|
|
sale_line = getattr(lot_s, 'sale_line', None)
|
|
sale_line_id = getattr(sale_line, 'id', None)
|
|
if sale_line and sale_line_id not in seen:
|
|
sale_lines.append(sale_line)
|
|
seen.add(sale_line_id)
|
|
|
|
if len(sale_lines) == 1:
|
|
return sale_lines[0]
|
|
return None
|
|
|
|
@classmethod
|
|
def _set_matched_sale_values(cls, values, sale_line):
|
|
if values and sale_line:
|
|
values['sale'] = sale_line.sale.id
|
|
values['sale_line'] = sale_line.id
|
|
|
|
@classmethod
|
|
def _fee_lot_segments(cls, lot, line, LotQt):
|
|
if getattr(lot, 'lot_type', None) == 'physic':
|
|
return [{
|
|
'quantity': None,
|
|
'share': None,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
lot_id = getattr(lot, 'id', None)
|
|
if not lot_id:
|
|
return [{
|
|
'quantity': None,
|
|
'share': None,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
domains = [[('lot_p', '=', lot_id), ('lot_quantity', '>', 0)]]
|
|
if getattr(lot, 'sale_line', None):
|
|
domains = [[('lot_s', '=', lot_id), ('lot_quantity', '>', 0)]]
|
|
|
|
lqts = []
|
|
seen = set()
|
|
for domain in domains:
|
|
for lqt in LotQt.search(domain):
|
|
lqt_id = cls._record_id(lqt)
|
|
if lqt_id in seen:
|
|
continue
|
|
seen.add(lqt_id)
|
|
lqts.append(lqt)
|
|
|
|
if not lqts:
|
|
return [{
|
|
'quantity': None,
|
|
'share': None,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}]
|
|
|
|
segment_line = getattr(lot, 'sale_line', None) or line
|
|
quantities = [cls._lotqt_quantity(lqt, segment_line) for lqt in lqts]
|
|
total = sum(quantities, Decimal(0))
|
|
|
|
segments = []
|
|
for lqt, quantity in zip(lqts, quantities):
|
|
segments.append({
|
|
'quantity': quantity,
|
|
'share': (quantity / total) if total else None,
|
|
'shipment_in': cls._lotqt_shipment_in_id(lqt),
|
|
})
|
|
return segments
|
|
|
|
@staticmethod
|
|
def _record_id(record):
|
|
return getattr(record, 'id', record)
|
|
|
|
@classmethod
|
|
def _unique_lots(cls, lots):
|
|
unique = []
|
|
seen = set()
|
|
for lot in lots:
|
|
lot_id = cls._record_id(lot)
|
|
if not lot or lot_id in seen:
|
|
continue
|
|
unique.append(lot)
|
|
seen.add(lot_id)
|
|
return tuple(unique)
|
|
|
|
@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 _value_key(cls, value):
|
|
return (
|
|
value.get('sale_line') or None,
|
|
None if value.get('sale_line') else value.get('line'),
|
|
value.get('lot'),
|
|
value.get('type'),
|
|
value.get('reference'),
|
|
value.get('counterparty'),
|
|
value.get('product'),
|
|
value.get('state'),
|
|
value.get('strategy'),
|
|
value.get('shipment_in'),
|
|
value.get('mtm_curve'),
|
|
)
|
|
|
|
@classmethod
|
|
def _dedupe_values(cls, values):
|
|
unique = []
|
|
seen = set()
|
|
for value in values:
|
|
key = cls._value_key(value)
|
|
if key in seen:
|
|
continue
|
|
unique.append(value)
|
|
seen.add(key)
|
|
return unique
|
|
|
|
@classmethod
|
|
def _snapshot_identity_domain(cls, value):
|
|
domain = [
|
|
('lot', '=', value.get('lot')),
|
|
('type', '=', value.get('type')),
|
|
('reference', '=', value.get('reference')),
|
|
('counterparty', '=', value.get('counterparty')),
|
|
('product', '=', value.get('product')),
|
|
('state', '=', value.get('state')),
|
|
('strategy', '=', value.get('strategy')),
|
|
('shipment_in', '=', value.get('shipment_in')),
|
|
('mtm_curve', '=', value.get('mtm_curve')),
|
|
]
|
|
if value.get('sale_line'):
|
|
domain.append(('sale_line', '=', value.get('sale_line')))
|
|
elif value.get('line'):
|
|
domain.append(('line', '=', value.get('line')))
|
|
elif value.get('sale'):
|
|
domain.append(('sale', '=', value.get('sale')))
|
|
elif value.get('purchase'):
|
|
domain.append(('purchase', '=', value.get('purchase')))
|
|
return domain
|
|
|
|
@classmethod
|
|
def _delete_existing_snapshot_values(cls, values):
|
|
if not values:
|
|
return
|
|
|
|
Date = Pool().get('ir.date')
|
|
Valuation = Pool().get('valuation.valuation')
|
|
ValuationLine = Pool().get('valuation.valuation.line')
|
|
seen = set()
|
|
|
|
for value in values:
|
|
if not value.get('lot') or not value.get('type'):
|
|
continue
|
|
key = cls._value_key(value)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
|
|
valuation_line_domain = cls._snapshot_identity_domain(value)
|
|
valuation_domain = list(valuation_line_domain)
|
|
valuation_domain.append(('date', '=', Date.today()))
|
|
|
|
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 _matched_purchase_lines_from_sale_line(cls, sale_line):
|
|
getter = getattr(sale_line, 'get_matched_lines', None)
|
|
if not callable(getter):
|
|
return []
|
|
|
|
purchase_lines = []
|
|
seen = set()
|
|
matched_lines = getter() or []
|
|
try:
|
|
iterator = iter(matched_lines)
|
|
except TypeError:
|
|
return []
|
|
for matched_line in iterator:
|
|
lot_p = getattr(matched_line, 'lot_p', None)
|
|
line = getattr(lot_p, 'line', None)
|
|
line_id = cls._record_id(line)
|
|
if line and line_id not in seen:
|
|
purchase_lines.append(line)
|
|
seen.add(line_id)
|
|
return purchase_lines
|
|
|
|
@classmethod
|
|
def _sale_line_is_unmatched(cls, sale_line):
|
|
return not cls._matched_purchase_lines_from_sale_line(sale_line)
|
|
|
|
@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,
|
|
shipment_in=None):
|
|
Date = Pool().get('ir.date')
|
|
|
|
values = {
|
|
'purchase': line.purchase.id,
|
|
'line': line.id,
|
|
'type': pnl_type,
|
|
'date': Date.today(),
|
|
'lot': lot.id,
|
|
'shipment_in': shipment_in or cls._lot_shipment_in_id(lot),
|
|
}
|
|
|
|
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,
|
|
'shipment_in': cls._lot_shipment_in_id(lot),
|
|
}
|
|
|
|
@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))
|
|
|
|
elif comp.price_source_type == 'fixed':
|
|
value = Decimal(comp.get_price(
|
|
scenario.valuation_date,
|
|
line.unit,
|
|
strategy.currency,
|
|
last=scenario.use_last_price
|
|
))
|
|
|
|
if comp.ratio:
|
|
value *= Decimal(comp.ratio) / Decimal(100)
|
|
|
|
total += value
|
|
|
|
return round(total, 4)
|
|
|
|
@staticmethod
|
|
def _supports_strategy_mtm(values):
|
|
return values and values.get('type') in {'pur. priced', 'sale priced'}
|
|
|
|
@classmethod
|
|
def _signed_strategy_mtm(cls, values, strategy, line):
|
|
mtm = strategy.get_mtm(line, values['quantity'])
|
|
if mtm is None:
|
|
return mtm
|
|
amount = values.get('amount')
|
|
if amount:
|
|
return abs(mtm) if amount > 0 else -abs(mtm)
|
|
if values.get('type') in {'pur. priced', 'pur. efp'}:
|
|
return -abs(mtm)
|
|
return abs(mtm)
|
|
|
|
@staticmethod
|
|
def _component_weight(component):
|
|
ratio = Decimal(component.ratio or 0)
|
|
return abs(ratio) if ratio else Decimal(100)
|
|
|
|
@classmethod
|
|
def _previous_curve_price(cls, curve, price_date, unit, currency):
|
|
PriceValue = Pool().get('price.price_value')
|
|
previous = PriceValue.search([
|
|
('price', '=', curve.id),
|
|
('price_date', '<', price_date),
|
|
], order=[('price_date', 'DESC')], limit=1)
|
|
if not previous:
|
|
return None
|
|
return Decimal(curve.get_price(
|
|
previous[0].price_date, unit, currency, last=False))
|
|
|
|
@classmethod
|
|
def _curve_component_price(cls, component, line, strategy):
|
|
scenario = strategy.scenario
|
|
value = Decimal(component.price_index.get_price(
|
|
scenario.valuation_date,
|
|
line.unit,
|
|
strategy.currency,
|
|
last=scenario.use_last_price))
|
|
previous = cls._previous_curve_price(
|
|
component.price_index,
|
|
scenario.valuation_date,
|
|
line.unit,
|
|
strategy.currency)
|
|
if component.ratio:
|
|
ratio = Decimal(component.ratio) / Decimal(100)
|
|
value *= ratio
|
|
if previous is not None:
|
|
previous *= ratio
|
|
return round(value, 4), (
|
|
round(previous, 4) if previous is not None else None)
|
|
|
|
@classmethod
|
|
def _split_value(cls, value, share, digits=2):
|
|
if value in (None, ''):
|
|
return value
|
|
return round(Decimal(value) * share, digits)
|
|
|
|
@classmethod
|
|
def _append_strategy_mtm_lines(cls, target, values, strategy, line):
|
|
components = list(strategy.components or [])
|
|
curve_components = [
|
|
component for component in components
|
|
if component.price_source_type == 'curve'
|
|
and component.price_index]
|
|
if not curve_components:
|
|
line_values = dict(values)
|
|
line_values['mtm_price'] = cls._get_strategy_mtm_price(
|
|
strategy, line)
|
|
line_values['amount_prev'] = None
|
|
line_values['mtm'] = cls._signed_strategy_mtm(
|
|
values, strategy, line)
|
|
line_values['strategy'] = strategy
|
|
target.append(line_values)
|
|
return
|
|
|
|
curve_component_ids = {id(component) for component in curve_components}
|
|
total_weight = sum(
|
|
cls._component_weight(component)
|
|
for component in components) or Decimal(100)
|
|
total_mtm_price = cls._get_strategy_mtm_price(strategy, line)
|
|
curve_mtm_price = Decimal(0)
|
|
signed_factor = Decimal(1)
|
|
amount = values.get('amount')
|
|
if amount and amount < 0:
|
|
signed_factor = Decimal(-1)
|
|
elif not amount and values.get('type') in {'pur. priced', 'pur. efp'}:
|
|
signed_factor = Decimal(-1)
|
|
|
|
for component in curve_components:
|
|
share = cls._component_weight(component) / total_weight
|
|
mtm_price, mtm_price_prev = cls._curve_component_price(
|
|
component, line, strategy)
|
|
curve_mtm_price += mtm_price
|
|
line_values = dict(values)
|
|
line_values['price'] = cls._split_value(
|
|
values.get('price'), share, digits=4)
|
|
line_values['amount'] = cls._split_value(
|
|
values.get('amount'), share)
|
|
line_values['base_amount'] = cls._split_value(
|
|
values.get('base_amount'), share)
|
|
line_values['mtm_price'] = mtm_price
|
|
line_values['mtm_price_prev'] = mtm_price_prev
|
|
line_values['mtm_curve'] = component.price_index.id
|
|
line_values['amount_prev'] = (
|
|
round(
|
|
mtm_price_prev * Decimal(values['quantity'])
|
|
* signed_factor, 2)
|
|
if mtm_price_prev is not None else None)
|
|
line_values['mtm'] = round(
|
|
mtm_price * Decimal(values['quantity']) * signed_factor, 2)
|
|
line_values['strategy'] = strategy
|
|
target.append(line_values)
|
|
|
|
residual_weight = sum(
|
|
cls._component_weight(component)
|
|
for component in components
|
|
if id(component) not in curve_component_ids)
|
|
if residual_weight:
|
|
share = residual_weight / total_weight
|
|
mtm_price = round(total_mtm_price - curve_mtm_price, 4)
|
|
line_values = dict(values)
|
|
line_values['price'] = cls._split_value(
|
|
values.get('price'), share, digits=4)
|
|
line_values['amount'] = cls._split_value(
|
|
values.get('amount'), share)
|
|
line_values['base_amount'] = cls._split_value(
|
|
values.get('base_amount'), share)
|
|
line_values['mtm_price'] = mtm_price
|
|
line_values['mtm_price_prev'] = None
|
|
line_values['mtm_curve'] = None
|
|
line_values['amount_prev'] = None
|
|
line_values['mtm'] = round(
|
|
mtm_price * Decimal(values['quantity']) * signed_factor, 2)
|
|
line_values['strategy'] = strategy
|
|
target.append(line_values)
|
|
|
|
@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), quantity=None, shipment_in=None):
|
|
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',
|
|
shipment_in=shipment_in,
|
|
)
|
|
|
|
qty = (Decimal(str(quantity)) if quantity is not None
|
|
else 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, quantity=None, shipment_in=None):
|
|
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,
|
|
shipment_in=shipment_in,
|
|
)
|
|
|
|
qty = (Decimal(str(quantity)) if quantity is not None
|
|
else 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 cls._valuation_lots(line):
|
|
for segment in cls._purchase_lot_segments(line, lot, LotQt):
|
|
cls._create_pnl_price_from_purchase_segment(
|
|
price_lines, line, lot, segment)
|
|
|
|
return price_lines
|
|
|
|
@classmethod
|
|
def _append_pnl_values(cls, price_lines, values, mtm_source):
|
|
if (values and getattr(mtm_source, 'mtm', None)
|
|
and cls._supports_strategy_mtm(values)):
|
|
for strat in mtm_source.mtm:
|
|
cls._append_strategy_mtm_lines(
|
|
price_lines, values, strat, mtm_source)
|
|
elif values:
|
|
price_lines.append(values)
|
|
|
|
@classmethod
|
|
def _create_pnl_price_from_purchase_segment(cls, price_lines, line, lot,
|
|
segment):
|
|
matched_sale_line = segment['sale_line']
|
|
quantity = segment['quantity']
|
|
shipment_in = segment.get('shipment_in')
|
|
|
|
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',
|
|
quantity=quantity,
|
|
shipment_in=shipment_in,
|
|
)
|
|
cls._set_matched_sale_values(values, matched_sale_line)
|
|
cls._append_pnl_values(price_lines, values, line)
|
|
else:
|
|
for pc in summaries:
|
|
values = cls._build_basis_pnl(
|
|
line=line, lot=lot, sale_line=None, pc=pc, sign=-1,
|
|
extra_price=premium_delta, quantity=quantity,
|
|
shipment_in=shipment_in)
|
|
cls._set_matched_sale_values(values, matched_sale_line)
|
|
cls._append_pnl_values(price_lines, values, line)
|
|
|
|
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}',
|
|
quantity=quantity,
|
|
shipment_in=shipment_in,
|
|
)
|
|
cls._set_matched_sale_values(values, matched_sale_line)
|
|
cls._append_pnl_values(price_lines, values, line)
|
|
|
|
sale_lot = segment['sale_lot']
|
|
if not sale_lot:
|
|
return
|
|
sale_line = sale_lot.sale_line
|
|
if not sale_line:
|
|
return
|
|
if cls._ignore_finished_open_lot(line, sale_lot):
|
|
return
|
|
if cls._ignore_finished_open_lot(sale_line, sale_lot):
|
|
return
|
|
|
|
if sale_line.price_type == 'basis':
|
|
premium_delta = cls._get_basis_premium_delta(sale_line)
|
|
for pc in sale_line.price_summary or []:
|
|
values = cls._build_basis_pnl(
|
|
line=line, lot=sale_lot, sale_line=sale_line, pc=pc,
|
|
sign=+1, extra_price=premium_delta,
|
|
quantity=segment.get('sale_quantity', quantity),
|
|
shipment_in=shipment_in)
|
|
cls._append_pnl_values(price_lines, values, sale_line)
|
|
|
|
elif sale_line.price_type in ('priced', 'efp'):
|
|
values = cls._build_simple_pnl(
|
|
line=line,
|
|
lot=sale_lot,
|
|
sale_line=sale_line,
|
|
price=sale_lot.lot_price_sale,
|
|
state='fixed' if sale_line.price_type == 'priced' else 'not fixed',
|
|
sign=+1,
|
|
pnl_type=f'sale {sale_line.price_type}',
|
|
quantity=segment.get('sale_quantity', quantity),
|
|
shipment_in=shipment_in,
|
|
)
|
|
cls._append_pnl_values(price_lines, values, sale_line)
|
|
|
|
@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 cls._valuation_lots(sale_line):
|
|
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:
|
|
cls._append_strategy_mtm_lines(
|
|
price_lines, values, strat, sale_line)
|
|
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:
|
|
cls._append_strategy_mtm_lines(
|
|
price_lines, values, strat, sale_line)
|
|
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:
|
|
cls._append_strategy_mtm_lines(
|
|
price_lines, values, strat, sale_line)
|
|
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 _fee_amount_or_zero(cls, fee):
|
|
amount = fee.get_amount() if hasattr(fee, 'get_amount') else fee.amount
|
|
return abs(Decimal(amount or 0))
|
|
|
|
@classmethod
|
|
def _fee_amount_for_lot_or_zero(cls, fee, lot):
|
|
if fee.mode == 'ppack' and getattr(lot, 'lot_qt', None):
|
|
return abs(round(
|
|
Decimal(fee.price or 0)
|
|
* Decimal(str(lot.lot_qt or 0)),
|
|
2))
|
|
if fee.mode == 'rate':
|
|
line = fee.line or getattr(fee, 'sale_line', None)
|
|
if line and line.estimated_date:
|
|
est_lines = [
|
|
dd for dd in line.estimated_date
|
|
if dd.trigger == 'bldate']
|
|
est_line = est_lines[0] if est_lines else None
|
|
if est_line and est_line.fin_int_delta:
|
|
factor = (
|
|
Decimal(fee.price or 0) / Decimal(100)
|
|
* Decimal(est_line.fin_int_delta) / Decimal(360))
|
|
quantity = Decimal(
|
|
lot.get_current_quantity_converted(0, fee.unit) or 0)
|
|
return abs(round(
|
|
factor * Decimal(line.unit_price or 0) * quantity,
|
|
2))
|
|
return Decimal(0)
|
|
return cls._fee_amount_or_zero(fee)
|
|
|
|
@classmethod
|
|
def _fee_segment_quantity(cls, fee, lot, segment):
|
|
quantity = segment.get('quantity')
|
|
if quantity is None:
|
|
qty = round(lot.get_current_quantity_converted(), 5)
|
|
if fee.mode == 'ppack' and getattr(lot, 'lot_qt', None):
|
|
qty = Decimal(str(lot.lot_qt or 0))
|
|
return qty
|
|
if fee.mode == 'ppack' and getattr(lot, 'lot_qt', None):
|
|
share = segment.get('share')
|
|
if share is not None:
|
|
return round(Decimal(str(lot.lot_qt or 0)) * share, 5)
|
|
return Decimal(str(lot.lot_qt or 0))
|
|
return round(Decimal(str(quantity)), 5)
|
|
|
|
@classmethod
|
|
def _fee_segment_amount(cls, fee, lot, qty, sign, segment):
|
|
share = segment.get('share')
|
|
if fee.mode == 'ppack':
|
|
price = fee.price
|
|
amount = cls._fee_amount_for_lot_or_zero(fee, lot)
|
|
if share is not None:
|
|
amount = round(amount * share, 2)
|
|
return price, amount * sign
|
|
if fee.mode == 'rate':
|
|
price = fee.price
|
|
line = fee.line or getattr(fee, 'sale_line', None)
|
|
if line and line.estimated_date:
|
|
est_lines = [
|
|
dd for dd in line.estimated_date
|
|
if dd.trigger == 'bldate']
|
|
est_line = est_lines[0] if est_lines else None
|
|
if est_line and est_line.fin_int_delta:
|
|
factor = (
|
|
Decimal(fee.price or 0) / Decimal(100)
|
|
* Decimal(est_line.fin_int_delta) / Decimal(360))
|
|
amount = abs(round(
|
|
factor * Decimal(line.unit_price or 0)
|
|
* Decimal(qty or 0), 2))
|
|
return price, amount * sign
|
|
return price, Decimal(0)
|
|
if fee.mode == 'lumpsum':
|
|
price = fee.price
|
|
amount = Decimal(price or 0)
|
|
if share is not None:
|
|
amount = round(amount * share, 2)
|
|
return price, amount * sign
|
|
price = Decimal(fee.get_price_per_qt())
|
|
return price, round(price * Decimal(qty or 0) * sign, 2)
|
|
|
|
@classmethod
|
|
def _fee_segments_for_fee(cls, fee, lot, line, LotQt):
|
|
segments = cls._fee_lot_segments(lot, line, LotQt)
|
|
shipment = getattr(fee, 'shipment_in', None)
|
|
shipment_id = cls._record_id(shipment) if shipment else None
|
|
if not shipment_id:
|
|
return segments
|
|
selected = [
|
|
segment for segment in segments
|
|
if segment.get('shipment_in') == shipment_id]
|
|
if selected:
|
|
return selected
|
|
return [{
|
|
'quantity': None,
|
|
'share': None,
|
|
'shipment_in': shipment_id,
|
|
}]
|
|
|
|
@staticmethod
|
|
def _check_fee_required_fields(fee):
|
|
if not getattr(fee, 'currency', None):
|
|
product = getattr(getattr(fee, 'product', None), 'rec_name', None)
|
|
product = product or getattr(getattr(fee, 'product', None), 'name', None)
|
|
if product:
|
|
raise UserError("Currency is required on fee %s." % product)
|
|
raise UserError("Currency is required on fee.")
|
|
|
|
@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')
|
|
LotQt = Pool().get('lot.qt')
|
|
#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 and s.lot_s.sale_line
|
|
and not cls._ignore_finished_open_lot(line, s.lot_s)
|
|
and not cls._ignore_finished_open_lot(s.lot_s.sale_line, s.lot_s)
|
|
)
|
|
all_lots = cls._unique_lots(cls._valuation_lots(line) + sale_open_lots)
|
|
for lot in all_lots:
|
|
if cls._ignore_empty_open_fee_lot(lot):
|
|
continue
|
|
if lot.sale_line and cls._ignore_finished_open_lot(lot.sale_line, lot):
|
|
continue
|
|
matched_sale_line = cls._get_matched_sale_line_from_purchase_lot(
|
|
lot, LotQt=LotQt)
|
|
fl = FeeLots.search([('lot', '=', lot.id)])
|
|
if not fl:
|
|
continue
|
|
|
|
fees = [
|
|
e.fee for e in fl
|
|
if e.fee and (
|
|
not hasattr(e.fee, 'is_effective_fee_lot')
|
|
or e.fee.is_effective_fee_lot(lot))
|
|
]
|
|
for sf in cls.group_fees_by_type_supplier(line, fees):
|
|
cls._check_fee_required_fields(sf)
|
|
sign = -1 if sf.p_r == 'pay' else 1
|
|
for segment in cls._fee_segments_for_fee(sf, lot, line, LotQt):
|
|
qty = cls._fee_segment_quantity(sf, lot, segment)
|
|
price, amount = cls._fee_segment_amount(
|
|
sf, lot, qty, sign, segment)
|
|
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': matched_sale_line.sale.id if matched_sale_line else None,
|
|
'sale_line': matched_sale_line.id if matched_sale_line else None,
|
|
'purchase': line.purchase.id,
|
|
'line': line.id,
|
|
'shipment_in': segment.get('shipment_in'),
|
|
'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')
|
|
LotQt = Pool().get('lot.qt')
|
|
|
|
for lot in cls._valuation_lots(sale_line):
|
|
if cls._ignore_empty_open_fee_lot(lot):
|
|
continue
|
|
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)
|
|
and (
|
|
not hasattr(e.fee, 'is_effective_fee_lot')
|
|
or e.fee.is_effective_fee_lot(lot))
|
|
]
|
|
for sf in cls.group_fees_by_type_supplier(sale_line, fees):
|
|
cls._check_fee_required_fields(sf)
|
|
sign = -1 if sf.p_r == 'pay' else 1
|
|
for segment in cls._fee_segments_for_fee(
|
|
sf, lot, sale_line, LotQt):
|
|
qty = cls._fee_segment_quantity(sf, lot, segment)
|
|
price, amount = cls._fee_segment_amount(
|
|
sf, lot, qty, sign, segment)
|
|
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,
|
|
'shipment_in': segment.get('shipment_in'),
|
|
'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)
|
|
values = cls._dedupe_values(values)
|
|
cls._delete_existing_snapshot_values(values)
|
|
|
|
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'):
|
|
matched_purchase_lines = cls._matched_purchase_lines_from_sale_line(
|
|
sale_line)
|
|
if matched_purchase_lines:
|
|
for line in matched_purchase_lines:
|
|
cls.generate(line, valuation_type=valuation_type)
|
|
return
|
|
|
|
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)
|
|
values = cls._dedupe_values(values)
|
|
cls._delete_existing_snapshot_values(values)
|
|
|
|
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 update_daily_snapshot(cls):
|
|
PurchaseLine = Pool().get('purchase.line')
|
|
SaleLine = Pool().get('sale.line')
|
|
|
|
purchase_lines = PurchaseLine.search([])
|
|
sale_lines = [
|
|
sale_line for sale_line in SaleLine.search([])
|
|
if cls._sale_line_is_unmatched(sale_line)
|
|
]
|
|
|
|
for line in purchase_lines:
|
|
cls.generate(line)
|
|
|
|
for sale_line in sale_lines:
|
|
cls.generate_from_sale_line(sale_line)
|
|
|
|
logger.info(
|
|
"Updated valuation snapshot for %s purchase line(s) and "
|
|
"%s unmatched sale line(s)",
|
|
len(purchase_lines),
|
|
len(sale_lines))
|
|
|
|
@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_amount_prev = fields.Numeric("Amount -1", 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_price_prev = fields.Numeric("Mtm Price -1", digits='r_unit')
|
|
r_mtm_curve = fields.Many2One('price.price', "Curve")
|
|
r_mtm = fields.Numeric("Mtm",digits='r_unit')
|
|
r_strategy = fields.Many2One('mtm.strategy',"Strategy")
|
|
r_lot = fields.Many2One('lot.lot',"Lot")
|
|
r_shipment_in = fields.Many2One('stock.shipment.in', "Shipment In")
|
|
|
|
@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.amount_prev).as_('r_amount_prev'),
|
|
Sum(val.base_amount).as_('r_base_amount'),
|
|
Sum(val.rate).as_('r_rate'),
|
|
Avg(val.mtm_price).as_('r_mtm_price'),
|
|
Avg(val.mtm_price_prev).as_('r_mtm_price_prev'),
|
|
val.mtm_curve.as_('r_mtm_curve'),
|
|
Sum(val.mtm).as_('r_mtm'),
|
|
Max(val.strategy).as_('r_strategy'),
|
|
Max(val.lot).as_('r_lot'),
|
|
Max(val.shipment_in).as_('r_shipment_in'),
|
|
where=wh,
|
|
group_by=[val.type, val.counterparty, val.state, val.mtm_curve])
|
|
|
|
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')
|
|
shipment_in = context.get('shipment_in')
|
|
if not valuation_date:
|
|
valuation_date = Date.today()
|
|
wh = (val.date == valuation_date)
|
|
if strategy:
|
|
wh &= (val.strategy == strategy)
|
|
if shipment_in:
|
|
wh &= (val.shipment_in == shipment_in)
|
|
|
|
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.amount_prev.as_('amount_prev'),
|
|
val.base_amount.as_('base_amount'),
|
|
val.rate.as_('rate'),
|
|
val.mtm_price.as_('mtm_price'),
|
|
val.mtm_price_prev.as_('mtm_price_prev'),
|
|
val.mtm_curve.as_('mtm_curve'),
|
|
val.mtm.as_('mtm'),
|
|
val.strategy.as_('strategy'),
|
|
val.lot.as_('lot'),
|
|
val.shipment_in.as_('shipment_in'),
|
|
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")
|
|
shipment_in = fields.Many2One('stock.shipment.in', "Shipment In")
|
|
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, {}
|