# This file is part of Tryton. The COPYRIGHT file at the top level of # this repository contains the full copyright notices and license terms. from trytond.model import fields from trytond.pool import Pool, PoolMeta from trytond.exceptions import UserError from trytond.pyson import Bool, Eval, Id from trytond.model import (ModelSQL, ModelView) from trytond.exceptions import UserError 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, StateTransition, StateView, Wizard from itertools import chain, groupby from operator import itemgetter import calendar import datetime import logging import re import zipfile from io import BytesIO from xml.etree import ElementTree from trytond.modules.purchase_trade.purchase import (TRIGGERS) from trytond.modules.purchase_trade.company_defaults import ( default_itsa_currency, default_itsa_price_calendar, default_itsa_price_fix_type, default_itsa_unit, is_itsa_company) logger = logging.getLogger(__name__) DAYTYPES = [ (None,''), ('before', 'Nb days before'), ('after', 'Nb days after'), ('first', 'First day'), ('last', 'Last day'), ('xth', 'Nth day'), ] WEEKDAY_MAP = { 'monday': 0, 'tuesday': 1, 'wednesday': 2, 'thursday': 3, 'friday': 4, 'saturday': 5, 'sunday': 6 } DAYS = [ (None,''), ('monday', 'Monday'), ('tuesday', 'Tuesday'), ('wednesday', 'Wednesday'), ('thursday', 'Thursday'), ('friday', 'Friday'), ('saturday', 'Saturday'), ('sunday', 'Sunday'), ] MTM_VALUATION_DATE_TYPES = [ ('fixed', 'Fixed Date'), ('today', 'Today'), ('month_start', 'Start of Month'), ('month_end', 'End of Month'), ('previous_month_end', 'End of Previous Month'), ('next_month_end', 'End of Next Month'), ('year_start', 'Start of Year'), ('year_end', 'End of Year'), ] class Estimated(ModelSQL, ModelView): "Estimated date" __name__ = 'pricing.estimated' trigger = fields.Selection(TRIGGERS,"Trigger") estimated_date = fields.Date("Estimated date") fin_int_delta = fields.Integer("Financing interests delta") class ImportPricesStart(ModelView): "Import Prices" __name__ = 'purchase_trade.import_prices.start' file_structure = fields.Selection([ ('historical', "Historical Prices"), ('historical_all_sources', "Historical Prices (All Sources)"), ('forward', "Forward Prices"), ], "Excel file structure", required=True) file_ = fields.Binary('Excel file', required=True, filename='filename') filename = fields.Char('Filename') create_missing_price_index = fields.Boolean( "Create price index if missing") overwrite_existing_price = fields.Boolean("Overwrite existing price") @staticmethod def default_file_structure(): return 'historical' class ImportPricesResult(ModelView): "Import Prices Result" __name__ = 'purchase_trade.import_prices.result' message = fields.Text("Results", readonly=True) class ImportPrices(Wizard): "Import Prices" __name__ = 'purchase_trade.import_prices' start = StateView( 'purchase_trade.import_prices.start', 'purchase_trade.import_prices_start_view_form', [ Button('Cancel', 'end', 'tryton-cancel'), Button('Import', 'import_', 'tryton-ok', default=True), ]) import_ = StateTransition() result = StateView( 'purchase_trade.import_prices.result', 'purchase_trade.import_prices_result_view_form', [ Button('OK', 'end', 'tryton-ok', default=True), ]) REQUIRED_COLUMNS = { 'priceindex': 'price_index', 'pricedate': 'price_date', 'highprice': 'high_price', 'lowprice': 'low_price', 'midprice': 'mid_price', 'openprice': 'open_price', 'pricevalue': 'price_value', } PRICE_INDEX_MAPPING_COLUMNS = ( 'excel_tab', 'original_price_index', 'source', 'column_destination', 'standard_price_index', 'index_description') ALL_SOURCES_VALUE_COLUMNS = { 'price_value', 'open_price', 'low_price', 'mid_price', 'high_price'} def transition_import_(self): rows = self._read_xlsx( self.start.file_, file_structure=self.start.file_structure) stats = self._import_rows( rows, create_missing_price_index=( self.start.create_missing_price_index), overwrite_existing_price=self.start.overwrite_existing_price) self._result_message = self._format_result(stats) return 'result' def default_result(self, fields): return { 'message': getattr( self, '_result_message', 'No import result was produced.'), } @classmethod def _import_rows( cls, rows, create_missing_price_index=False, overwrite_existing_price=False): Price = Pool().get('price.price') PriceValue = Pool().get('price.price_value') stats = { 'created_indexes': [], 'imported': [], 'updated': [], 'skipped': [], 'errors': [], } for index, row in enumerate(rows, start=2): row_number = row.get('_row_number', index) price_index = (row.get('price_index') or '').strip() try: price_date = cls._as_date(row.get('price_date')) if not price_index: stats['skipped'].append( cls._result_line( row_number, '', None, 'missing price_index')) continue if not price_date: stats['skipped'].append( cls._result_line( row_number, price_index, None, 'missing price_date')) continue prices = Price.search( [('price_index', '=', price_index)], limit=1) if prices: price = prices[0] elif create_missing_price_index: price, = Price.create([ cls._price_index_values(price_index, row)]) stats['created_indexes'].append( cls._result_line( row_number, price_index, None, 'price index created')) else: stats['skipped'].append( cls._result_line( row_number, price_index, None, 'price_index missing')) continue values = cls._price_value_values(price, row, price_date) existing = PriceValue.search([ ('price', '=', price.id), ('price_date', '=', price_date), ], limit=1) if existing: if overwrite_existing_price: PriceValue.write(existing, values) stats['updated'].append( cls._result_line( row_number, price_index, price_date, cls._price_summary(values))) else: stats['skipped'].append(cls._result_line( row_number, price_index, price_date, 'price_date already exists')) continue PriceValue.create([values]) stats['imported'].append( cls._result_line( row_number, price_index, price_date, cls._price_summary(values))) except Exception as exception: stats['errors'].append( cls._result_line( row_number, price_index, row.get('price_date'), str(exception))) continue return stats @classmethod def _price_index_values(cls, price_index, row=None): row = row or {} values = { 'price_index': price_index, 'price_desc': row.get('_price_desc') or price_index, 'price_curve_type': 'future', } references = [ ('price_type', 'price.fixtype', [('name', '=', 'Market price')]), ('price_currency', 'currency.currency', [('name', '=', 'USD')]), ('price_calendar', 'price.calendar', [('name', '=', 'Argus EU')]), ('price_unit', 'product.uom', [('name', '=', 'Mt')]), ] source = row.get('_source') if source: references.append( ('price_area', 'price.area', [('name', '=', source)])) for field, model_name, domain in references: record = cls._first_record(model_name, domain) if record: values[field] = record.id period = cls._period_from_price_index(price_index) if period: values['price_period'] = period.id return values @classmethod def _period_from_price_index(cls, price_index): match = re.search(r'(?' price_date = item['price_date'] if isinstance(price_date, datetime.date): price_date = price_date.isoformat() if price_date: label = '%s / %s' % (label, price_date) return 'Row %(row)s - ' % item + '%s: %s' % (label, item['detail']) @staticmethod def _price_summary(values): parts = [] for name in ( 'price_value', 'open_price', 'low_price', 'mid_price', 'high_price'): value = values.get(name) if value is not None: parts.append('%s=%s' % (name, value)) return ', '.join(parts) or 'price imported' @classmethod def _read_xlsx(cls, data, file_structure='historical'): try: with zipfile.ZipFile(BytesIO(data)) as workbook: shared_strings = cls._read_shared_strings(workbook) if file_structure == 'historical_all_sources': return cls._read_historical_all_sources_price_rows( workbook, shared_strings) sheet_name = cls._first_sheet_name(workbook) sheet = ElementTree.fromstring(workbook.read(sheet_name)) except (KeyError, zipfile.BadZipFile, ElementTree.ParseError): raise UserError("The selected file is not a valid Excel .xlsx file.") ns = {'s': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} rows = sheet.findall('.//s:sheetData/s:row', ns) if not rows: return [] if file_structure == 'forward': return cls._read_forward_price_rows(rows, shared_strings, ns) return cls._read_historical_price_rows(rows, shared_strings, ns) @classmethod def _read_historical_all_sources_price_rows(cls, workbook, shared_strings): ns = {'s': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} sheet_rows = cls._workbook_sheet_rows(workbook, ns) mapping = cls._price_index_mapping() return cls._read_historical_all_sources_rows( sheet_rows, mapping, shared_strings, ns) @classmethod def _read_historical_all_sources_rows( cls, sheet_rows, mapping, shared_strings, ns): result_by_key = {} mapping_by_tab = {} for mapping_row in mapping: tab = mapping_row.get('excel_tab') destination = mapping_row.get('column_destination') if not tab or destination not in cls.ALL_SOURCES_VALUE_COLUMNS: continue mapping_by_tab.setdefault(tab, []).append(mapping_row) for tab, tab_mapping in mapping_by_tab.items(): sheet_name = cls._matching_sheet_name(sheet_rows, tab) if not sheet_name: continue rows = sheet_rows[sheet_name] header_row_number, mapped_columns = cls._mapped_columns( rows, tab_mapping, shared_strings, ns) if not mapped_columns: continue date_column = cls._date_column( rows[header_row_number - 1], mapped_columns, shared_strings, ns) for excel_row_number, sheet_row in enumerate( rows[header_row_number:], start=header_row_number + 1): row_values = cls._row_values(sheet_row, shared_strings, ns) price_date = row_values.get(date_column) if price_date in (None, ''): continue for column, mapping_row in mapped_columns.items(): price_value = row_values.get(column) if price_value in (None, ''): continue price_index = ( mapping_row.get('standard_price_index') or '').strip() if not price_index: continue key = (price_index, price_date, excel_row_number) result = result_by_key.setdefault(key, { '_row_number': excel_row_number, '_source': mapping_row.get('source'), '_price_desc': mapping_row.get('index_description'), 'price_index': price_index, 'price_date': price_date, }) result[mapping_row['column_destination']] = price_value if (mapping_row['column_destination'] == 'mid_price' and not result.get('price_value')): result['price_value'] = price_value return list(result_by_key.values()) @classmethod def _price_index_mapping(cls): cursor = Transaction().connection.cursor() columns = ', '.join(cls.PRICE_INDEX_MAPPING_COLUMNS) cursor.execute( 'SELECT %s FROM public."ext_H2SO4_PriceIndexMapping"' % columns) return [ dict(zip(cls.PRICE_INDEX_MAPPING_COLUMNS, row)) for row in cursor.fetchall()] @classmethod def _mapped_columns(cls, rows, mapping, shared_strings, ns): mapping_by_header = {} for mapping_row in mapping: header = cls._normalize_mapping_value( mapping_row.get('original_price_index')) mapping_by_header[header] = mapping_row hits_by_row = {} columns_by_row = {} for row_number, sheet_row in enumerate(rows, start=1): for cell in sheet_row.findall('s:c', ns): header = cls._normalize_mapping_value( cls._cell_value(cell, shared_strings)) mapping_row = mapping_by_header.get(header) if not mapping_row: continue column = cls._cell_column_index(cell.get('r')) hits_by_row[row_number] = hits_by_row.get(row_number, 0) + 1 columns_by_row.setdefault(row_number, {})[column] = mapping_row if not hits_by_row: return None, {} header_row_number = max(hits_by_row, key=hits_by_row.get) return header_row_number, columns_by_row[header_row_number] @classmethod def _date_column(cls, header_row, mapped_columns, shared_strings, ns): for cell in header_row.findall('s:c', ns): if cls._normalize_header( cls._cell_value(cell, shared_strings)) == 'date': return cls._cell_column_index(cell.get('r')) return min(mapped_columns) - 1 @classmethod def _row_values(cls, sheet_row, shared_strings, ns): values = {} for cell in sheet_row.findall('s:c', ns): values[cls._cell_column_index(cell.get('r'))] = cls._cell_value( cell, shared_strings) return values @classmethod def _workbook_sheet_rows(cls, workbook, ns): sheet_rows = {} for sheet_name, sheet_path in cls._worksheet_paths(workbook): sheet = ElementTree.fromstring(workbook.read(sheet_path)) sheet_rows[sheet_name] = sheet.findall('.//s:sheetData/s:row', ns) return sheet_rows @classmethod def _worksheet_paths(cls, workbook): workbook_xml = ElementTree.fromstring(workbook.read('xl/workbook.xml')) rels_xml = ElementTree.fromstring( workbook.read('xl/_rels/workbook.xml.rels')) wb_ns = { 's': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main', 'r': ( 'http://schemas.openxmlformats.org/officeDocument/2006/' 'relationships'), } rel_ns = { 'r': ( 'http://schemas.openxmlformats.org/package/2006/' 'relationships'), } rels = { rel.get('Id'): rel.get('Target') for rel in rels_xml.findall('r:Relationship', rel_ns)} for sheet in workbook_xml.findall('.//s:sheet', wb_ns): rel_id = sheet.get( '{http://schemas.openxmlformats.org/officeDocument/2006/' 'relationships}id') target = rels[rel_id] if not target.startswith('/'): target = 'xl/' + target yield sheet.get('name'), target.lstrip('/') @classmethod def _matching_sheet_name(cls, sheet_rows, mapping_tab): normalized_tab = cls._normalize_mapping_value(mapping_tab) for sheet_name in sheet_rows: if cls._normalize_mapping_value(sheet_name) == normalized_tab: return sheet_name return None @classmethod def _read_historical_price_rows(cls, rows, shared_strings, ns): headers = {} for cell in rows[0].findall('s:c', ns): index = cls._cell_column_index(cell.get('r')) header = cls._cell_value(cell, shared_strings) normalized = cls._normalize_header(header) if normalized in cls.REQUIRED_COLUMNS: headers[index] = cls.REQUIRED_COLUMNS[normalized] missing = set(cls.REQUIRED_COLUMNS.values()) - set(headers.values()) if missing: raise UserError( "Missing columns in Excel file: %s" % ', '.join(sorted(missing))) result = [] for sheet_row in rows[1:]: values = {} for cell in sheet_row.findall('s:c', ns): index = cls._cell_column_index(cell.get('r')) field = headers.get(index) if field: values[field] = cls._cell_value(cell, shared_strings) if any(v not in (None, '') for v in values.values()): result.append(values) return result @classmethod def _read_forward_price_rows(cls, rows, shared_strings, ns): headers = {} month_terms = {} for cell in rows[0].findall('s:c', ns): index = cls._cell_column_index(cell.get('r')) header = cls._cell_value(cell, shared_strings) normalized = cls._normalize_header(header) if index == 1 and normalized == 'priceindex': headers[index] = 'price_index' elif index == 2 and normalized == 'pricedate': headers[index] = 'price_date' elif index and index >= 3 and header not in (None, ''): month_terms[index] = cls._as_month_term(header) missing = [] if headers.get(1) != 'price_index': missing.append('price_index') if headers.get(2) != 'price_date': missing.append('price_date') if missing: raise UserError( "Missing columns in Excel file: %s" % ', '.join(missing)) if not month_terms: raise UserError( "Missing month term columns in Excel file.") result = [] for excel_row_number, sheet_row in enumerate(rows[1:], start=2): row_values = {} for cell in sheet_row.findall('s:c', ns): index = cls._cell_column_index(cell.get('r')) row_values[index] = cls._cell_value(cell, shared_strings) price_index = (row_values.get(1) or '').strip() price_date = row_values.get(2) if not price_index and price_date in (None, ''): continue for index, month_term in sorted(month_terms.items()): price_value = row_values.get(index) if price_value in (None, ''): continue result.append({ '_row_number': excel_row_number, 'price_index': ( '%s %s' % (price_index, month_term) if price_index else ''), 'price_date': price_date, 'high_price': price_value, 'low_price': price_value, 'mid_price': price_value, 'open_price': price_value, 'price_value': price_value, }) return result @staticmethod def _read_shared_strings(workbook): try: content = workbook.read('xl/sharedStrings.xml') except KeyError: return [] root = ElementTree.fromstring(content) ns = {'s': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} strings = [] for item in root.findall('s:si', ns): strings.append(''.join( text.text or '' for text in item.findall('.//s:t', ns))) return strings @staticmethod def _first_sheet_name(workbook): workbook_xml = ElementTree.fromstring(workbook.read('xl/workbook.xml')) rels_xml = ElementTree.fromstring( workbook.read('xl/_rels/workbook.xml.rels')) wb_ns = { 's': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main', 'r': ( 'http://schemas.openxmlformats.org/officeDocument/2006/' 'relationships'), } rel_ns = { 'r': ( 'http://schemas.openxmlformats.org/package/2006/' 'relationships'), } sheet = workbook_xml.find('.//s:sheet', wb_ns) rel_id = sheet.get( '{http://schemas.openxmlformats.org/officeDocument/2006/' 'relationships}id') for rel in rels_xml.findall('r:Relationship', rel_ns): if rel.get('Id') == rel_id: target = rel.get('Target') if not target.startswith('/'): target = 'xl/' + target return target.lstrip('/') raise KeyError('No worksheet found') @staticmethod def _cell_column_index(reference): match = re.match(r'([A-Z]+)', reference or '') if not match: return None index = 0 for char in match.group(1): index = index * 26 + ord(char) - ord('A') + 1 return index @classmethod def _cell_value(cls, cell, shared_strings): ns = {'s': 'http://schemas.openxmlformats.org/spreadsheetml/2006/main'} if cell.get('t') == 'inlineStr': text = cell.find('.//s:t', ns) return text.text if text is not None else '' value = cell.find('s:v', ns) if value is None: return '' if cell.get('t') == 's': return shared_strings[int(value.text)] return value.text @staticmethod def _normalize_header(value): return re.sub(r'[^a-z0-9]', '', (value or '').strip().lower()) @staticmethod def _normalize_mapping_value(value): return re.sub(r'\s+', ' ', (value or '').strip()).lower() @staticmethod def _as_float(value): if value in (None, ''): return None return float(str(value).replace(',', '.')) @staticmethod def _as_date(value): if isinstance(value, datetime.date): return value if value in (None, ''): return None text = str(value).strip() if re.match(r'^\d+(\.\d+)?$', text): return ( datetime.date(1899, 12, 30) + datetime.timedelta(days=int(float(text)))) for fmt in ('%Y-%m-%d', '%d/%m/%Y', '%m/%d/%Y'): try: return datetime.datetime.strptime(text, fmt).date() except ValueError: pass raise UserError("Invalid price_date: %s" % text) @classmethod def _as_month_term(cls, value): if isinstance(value, datetime.date): return value.strftime('%Y-%m') text = str(value).strip() match = re.match(r'^(20\d{2})[-_/\. ](0[1-9]|1[0-2])$', text) if match: return '%s-%s' % (match.group(1), match.group(2)) if re.match(r'^\d+(\.\d+)?$', text): return cls._as_date(text).strftime('%Y-%m') raise UserError("Invalid month term: %s" % text) class MtmScenario(ModelSQL, ModelView): "MtM Scenario" __name__ = 'mtm.scenario' name = fields.Char("Scenario", required=True) valuation_date_type = fields.Selection( MTM_VALUATION_DATE_TYPES, "Valuation Date Type") valuation_date = fields.Date( "Valuation Date", states={ 'required': Eval('valuation_date_type') == 'fixed', }, depends=['valuation_date_type']) resolved_valuation_date = fields.Function( fields.Date("Resolved Valuation Date"), 'get_resolved_valuation_date') use_last_price = fields.Boolean("Use Last Available Price") calendar = fields.Many2One( 'price.calendar', "Calendar" ) @staticmethod def default_valuation_date_type(): return 'fixed' @classmethod def default_calendar(cls): if is_itsa_company(): calendar = default_itsa_price_calendar() if calendar: return calendar def get_resolved_valuation_date(self, name=None): return self.get_valuation_date() def get_valuation_date(self): date_type = self.valuation_date_type or 'fixed' if date_type == 'fixed': return self.valuation_date today = Pool().get('ir.date').today() if date_type == 'today': return today if date_type == 'month_start': return datetime.date(today.year, today.month, 1) if date_type == 'month_end': return datetime.date( today.year, today.month, calendar.monthrange(today.year, today.month)[1]) if date_type == 'previous_month_end': return ( datetime.date(today.year, today.month, 1) - datetime.timedelta(days=1)) if date_type == 'next_month_end': year = today.year + (today.month // 12) month = (today.month % 12) + 1 return datetime.date( year, month, calendar.monthrange(year, month)[1]) if date_type == 'year_start': return datetime.date(today.year, 1, 1) if date_type == 'year_end': return datetime.date(today.year, 12, 31) return self.valuation_date class MtmStrategy(ModelSQL, ModelView): "Mark to Market Strategy" __name__ = 'mtm.strategy' name = fields.Char("Name", required=True) active = fields.Boolean("Active") scenario = fields.Many2One( 'mtm.scenario', "Scenario", required=True ) currency = fields.Many2One( 'currency.currency', "Valuation Currency" ) components = fields.One2Many( 'pricing.component', 'strategy', "Components" ) @classmethod def default_active(cls): return True @classmethod def default_currency(cls): if is_itsa_company(): currency = default_itsa_currency() if currency: return currency @staticmethod def _scenario_valuation_date(scenario): get_valuation_date = getattr( type(scenario), 'get_valuation_date', None) if callable(get_valuation_date): return scenario.get_valuation_date() return scenario.valuation_date def get_mtm(self,line,qty): pool = Pool() Currency = pool.get('currency.currency') total = Decimal(0) scenario = self.scenario dt = self._scenario_valuation_date(scenario) for comp in self.components: value = Decimal(0) if comp.price_source_type == 'curve' and comp.price_index: value = Decimal( comp.price_index.get_price( dt, line.unit, self.currency, relative_last=scenario.use_last_price ) ) elif comp.price_source_type == 'matrix' and comp.price_matrix: value = self._get_matrix_price(comp, line, dt) elif comp.price_source_type == 'fixed': value = Decimal(comp.get_price( dt, line.unit, self.currency, relative_last=scenario.use_last_price )) if comp.ratio: value *= Decimal(comp.ratio) / Decimal(100) total += value * qty return Decimal(str(total)).quantize(Decimal("0.01")) def _get_matrix_price(self, comp, line, dt): MatrixLine = Pool().get('price.matrix.line') domain = [ ('matrix', '=', comp.price_matrix.id), ] if line: domain += [ ('origin', '=', line.purchase.from_location), ('destination', '=', line.purchase.to_location), ] lines = MatrixLine.search(domain) if lines: return Decimal(lines[0].price_value) return Decimal(0) def run_daily_mtm(): Strategy = Pool().get('mtm.strategy') Snapshot = Pool().get('mtm.snapshot') for strat in Strategy.search([('active', '=', True)]): amount = strat.compute_mtm() valuation_date = strat._scenario_valuation_date(strat.scenario) Snapshot.create([{ 'strategy': strat.id, 'valuation_date': valuation_date, 'amount': amount, 'currency': strat.currency.id, }]) class Mtm(ModelSQL, ModelView): "MtM Component" __name__ = 'mtm.component' strategy = fields.Many2One( 'mtm.strategy', "Strategy", required=True, ondelete='CASCADE' ) name = fields.Char("Component", required=True) component_type = fields.Selection([ ('commodity', 'Commodity'), ('freight', 'Freight'), ('quality', 'Quality'), ('fx', 'FX'), ('storage', 'Storage'), ('other', 'Other'), ], "Type", required=True) fix_type = fields.Many2One('price.fixtype', "Fixation Type") price_source_type = fields.Selection([ ('curve', 'Curve'), ('matrix', 'Matrix'), ('manual', 'Manual'), ], "Price Source", required=True) price_index = fields.Many2One('price.price', "Price Curve") price_matrix = fields.Many2One('price.matrix', "Price Matrix") ratio = fields.Numeric("Ratio / %", digits=(16, 6)) manual_price = fields.Numeric( "Manual Price", digits=(16, 6), help="Price set manually if price_source_type is 'manual'" ) currency = fields.Many2One('currency.currency', "Currency") def get_cur(self, name=None): if self.price_index: return self.price_index.price_currency if self.price_matrix: return self.price_matrix.currency if self.currency: return self.currency return None @fields.depends('price_index','price_matrix') def on_change_with_currency(self): return self.get_cur() @classmethod def default_currency(cls): if is_itsa_company(): currency = default_itsa_currency() if currency: return currency @classmethod def default_fix_type(cls): if is_itsa_company(): fix_type = default_itsa_price_fix_type() if fix_type: return fix_type @classmethod def default_ratio(cls): if is_itsa_company(): return Decimal('100') class PriceMatrix(ModelSQL, ModelView): "Price Matrix" __name__ = 'price.matrix' name = fields.Char("Name", required=True) matrix_type = fields.Selection([ ('freight', 'Freight'), ('location', 'Location Spread'), ('quality', 'Quality'), ('storage', 'Storage'), ('other', 'Other'), ], "Matrix Type", required=True) unit = fields.Many2One('product.uom', "Unit") currency = fields.Many2One('currency.currency', "Currency") calendar = fields.Many2One( 'price.calendar', "Calendar" ) valid_from = fields.Date("Valid From") valid_to = fields.Date("Valid To") lines = fields.One2Many( 'price.matrix.line', 'matrix', "Lines" ) @classmethod def default_currency(cls): if is_itsa_company(): currency = default_itsa_currency() if currency: return currency @classmethod def default_unit(cls): if is_itsa_company(): unit = default_itsa_unit() if unit: return unit class PriceMatrixLine(ModelSQL, ModelView): "Price Matrix Line" __name__ = 'price.matrix.line' matrix = fields.Many2One( 'price.matrix', "Matrix", required=True, ondelete='CASCADE' ) origin = fields.Many2One('stock.location', "Origin") destination = fields.Many2One('stock.location', "Destination") product = fields.Many2One('product.product', "Product") quality = fields.Many2One('product.category', "Quality") price_value = fields.Numeric("Price", digits=(16, 6)) class MtmSnapshot(ModelSQL, ModelView): "MtM Snapshot" __name__ = 'mtm.snapshot' strategy = fields.Many2One( 'mtm.strategy', "Strategy", required=True, ondelete='CASCADE' ) valuation_date = fields.Date("Valuation Date", required=True) amount = fields.Numeric("MtM Amount", digits=(16, 6)) currency = fields.Many2One('currency.currency', "Currency") created_at = fields.DateTime("Created At") class Component(ModelSQL, ModelView): "Component" __name__ = 'pricing.component' strategy = fields.Many2One( 'mtm.strategy', "Strategy", required=False, ondelete='CASCADE' ) price_source_type = fields.Selection([ ('curve', 'Curve'), ('matrix', 'Matrix'), ('fixed', 'Fixed'), ], "Price Source", required=True) fix_type = fields.Many2One('price.fixtype',"Fixation type") ratio = fields.Numeric("%",digits=(16,7)) price_index = fields.Many2One( 'price.price', "Curve", states={ 'readonly': Eval('price_source_type') != 'curve', 'required': Eval('price_source_type') == 'curve', }, depends=['price_source_type']) price_matrix = fields.Many2One( 'price.matrix', "Price Matrix", states={ 'readonly': Eval('price_source_type') != 'matrix', 'required': Eval('price_source_type') == 'matrix', }, depends=['price_source_type']) fixed_price = fields.Numeric( "Fixed Price", digits=(16, 6), states={ 'readonly': Eval('price_source_type') != 'fixed', 'required': Eval('price_source_type') == 'fixed', }, depends=['price_source_type']) fixed_currency = fields.Many2One('currency.currency', "Fixed Curr.") currency = fields.Function( fields.Many2One( 'currency.currency', "Curr.", states={ 'required': Eval('price_source_type') == 'fixed', }, depends=['price_source_type']), 'get_cur', setter='set_cur') auto = fields.Boolean("Auto") fallback = fields.Boolean("Fallback") calendar = fields.Many2One('price.calendar',"Calendar") nbdays = fields.Function(fields.Integer("Nb days"),'get_nbdays') triggers = fields.One2Many('pricing.trigger','component',"Period rules") pricing_date = fields.Date("Pricing date max") def get_rec_name(self, name=None): if self.price_index: return '[' + self.fix_type.name + '] ' + self.price_index.price_index if self.price_matrix: return '[' + self.fix_type.name + '] ' + self.price_matrix.name if self.price_source_type == 'fixed': return '[' + self.fix_type.name + '] Fixed' else: return '[' + self.fix_type.name + '] ' def get_cur(self, name=None): if self.price_index: PI = Pool().get('price.price') pi = PI(self.price_index) return pi.price_currency if self.price_matrix: return self.price_matrix.currency return self.fixed_currency @fields.depends( 'price_source_type', 'price_index', 'price_matrix', 'fixed_currency') def on_change_with_currency(self): if self.price_source_type == 'curve' and self.price_index: return self.get_cur() if self.price_source_type == 'matrix' and self.price_matrix: return self.get_cur() return self.fixed_currency @classmethod def default_fixed_currency(cls): if is_itsa_company(): currency = default_itsa_currency() if currency: return currency @classmethod def default_fix_type(cls): if is_itsa_company(): fix_type = default_itsa_price_fix_type() if fix_type: return fix_type @classmethod def default_ratio(cls): if is_itsa_company(): return Decimal('100') @classmethod def default_calendar(cls): if is_itsa_company(): calendar = default_itsa_price_calendar() if calendar: return calendar @classmethod def set_cur(cls, components, name, value): cls.write(components, {'fixed_currency': value}) def get_calendar(self): if self.calendar: return self.calendar if self.price_matrix: return self.price_matrix.calendar @staticmethod def _record_id(record): return getattr(record, 'id', record) @classmethod def _trigger_values(cls, trigger): return { 'pricing_period': cls._record_id( getattr(trigger, 'pricing_period', None)), 'from_p': getattr(trigger, 'from_p', None), 'to_p': getattr(trigger, 'to_p', None), 'average': getattr(trigger, 'average', None), 'last': getattr(trigger, 'last', None), 'relative_last': getattr(trigger, 'relative_last', None), 'application_period': cls._record_id( getattr(trigger, 'application_period', None)), 'from_a': getattr(trigger, 'from_a', None), 'to_a': getattr(trigger, 'to_a', None), } @classmethod def _copy_trigger_values(cls, component): return [ cls._trigger_values(trigger) for trigger in getattr(component, 'triggers', [])] @classmethod def _first_component_domain(cls, line=None, sale_line=None, exclude=None): domain = [] if line: domain.append(('line', '=', cls._record_id(line))) elif sale_line: domain.append(('sale_line', '=', cls._record_id(sale_line))) else: return None if exclude: domain.append(('id', '!=', cls._record_id(exclude))) return domain @classmethod def _get_first_component(cls, line=None, sale_line=None, exclude=None): domain = cls._first_component_domain(line, sale_line, exclude) if not domain: return None components = cls.search(domain, order=[('id', 'ASC')], limit=1) return components[0] if components else None @classmethod def _default_trigger_values(cls, line=None, sale_line=None, exclude=None): component = cls._get_first_component(line, sale_line, exclude) if not component: return [] return cls._copy_trigger_values(component) @classmethod def default_triggers(cls): context = Transaction().context return cls._default_trigger_values( line=context.get('default_line'), sale_line=context.get('default_sale_line')) def _inherit_first_component_triggers(self): if getattr(self, 'triggers', None): return values = self._default_trigger_values( line=getattr(self, 'line', None), sale_line=getattr(self, 'sale_line', None), exclude=getattr(self, 'id', None)) if values: Trigger = Pool().get('pricing.trigger') self.triggers = [Trigger(**value) for value in values] @fields.depends('line', 'sale_line', 'triggers') def on_change_line(self): self._inherit_first_component_triggers() @fields.depends('line', 'sale_line', 'triggers') def on_change_sale_line(self): self._inherit_first_component_triggers() @classmethod def create(cls, vlist): vlist = [values.copy() for values in vlist] for values in vlist: if values.get('triggers'): continue trigger_values = cls._default_trigger_values( line=values.get('line'), sale_line=values.get('sale_line')) if trigger_values: values['triggers'] = [('create', trigger_values)] return super(Component, cls).create(vlist) @classmethod def _matches_optional_field(cls, matrix_line, field_name, target): value = getattr(matrix_line, field_name, None) if not value: return True, 0 if target and cls._record_id(value) == cls._record_id(target): return True, 1 return False, 0 def _matrix_valid_on(self, price_date): if not self.price_matrix: return False if hasattr(price_date, 'date'): price_date = price_date.date() valid_from = getattr(self.price_matrix, 'valid_from', None) valid_to = getattr(self.price_matrix, 'valid_to', None) if valid_from and price_date < valid_from: return False if valid_to and price_date > valid_to: return False return True def _get_matrix_owner_values(self): line = getattr(self, 'line', None) or getattr(self, 'sale_line', None) document = getattr(line, 'purchase', None) or getattr(line, 'sale', None) return { 'origin': getattr(document, 'from_location', None), 'destination': getattr(document, 'to_location', None), 'product': getattr(line, 'product', None), 'quality': getattr(line, 'quality', None), } def _get_matrix_line(self): MatrixLine = Pool().get('price.matrix.line') lines = MatrixLine.search([('matrix', '=', self.price_matrix.id)]) owner_values = self._get_matrix_owner_values() best_line = None best_score = -1 for line in lines: score = 0 for field_name, target in owner_values.items(): matches, field_score = self._matches_optional_field( line, field_name, target) if not matches: break score += field_score else: if score > best_score: best_line = line best_score = score return best_line def _convert_matrix_price(self, price, unit, currency): price_qt = Decimal(str(price or 0)) matrix_unit = getattr(self.price_matrix, 'unit', None) matrix_currency = getattr(self.price_matrix, 'currency', None) if matrix_unit and unit: Uom = Pool().get('product.uom') price_qt *= Decimal(str( Uom.compute_qty(unit, float(1), matrix_unit))) if matrix_currency and currency and currency != matrix_currency: Currency = Pool().get('currency.currency') rates = Currency._get_rate([matrix_currency]) rate = rates.get(matrix_currency.id) if rate: price_qt *= Decimal(str(rate)) return round(price_qt, 4) def _convert_fixed_price(self, price, currency): price_qt = Decimal(str(price or 0)) price_currency = getattr(self, 'fixed_currency', None) if price_currency and currency and currency != price_currency: Currency = Pool().get('currency.currency') rates = Currency._get_rate([price_currency]) rate = rates.get(price_currency.id) if rate: price_qt *= Decimal(str(rate)) return round(price_qt, 4) def get_price( self, price_date, unit, currency, last=False, relative_last=False): if self.price_source_type == 'curve' and self.price_index: PI = Pool().get('price.price') pi = PI(self.price_index) return pi.get_price( price_date, unit, currency, last, relative_last) if self.price_source_type == 'matrix' and self.price_matrix: if not self._matrix_valid_on(price_date): return Decimal(0) line = self._get_matrix_line() if line: return self._convert_matrix_price( line.price_value, unit, currency) if self.price_source_type == 'fixed': return self._convert_fixed_price(self.fixed_price, currency) return Decimal(0) def get_nbdays(self, name): days = 0 if self.triggers: for t in self.triggers: l,l2 = t.getApplicationListDates(self.get_calendar()) days += len(l) return days @classmethod def delete(cls, components): for cp in components: Pricing = Pool().get('pricing.pricing') pricings = Pricing.search(['price_component','=',cp.id]) if pricings: Pricing.delete(pricings) super(Component, cls).delete(components) class Pricing(ModelSQL,ModelView): "Pricing" __name__ = 'pricing.pricing' pricing_date = fields.Date("Date") price_component = fields.Many2One( 'pricing.component', "Component", domain=[('id', 'in', Eval('available_components', []))], depends=['available_components'])#, ondelete='CASCADE') available_components = fields.Function( fields.One2Many('pricing.component', '', "Available Components"), 'on_change_with_available_components', setter='set_available_components') quantity = fields.Numeric("Qt",digits='unit') settl_price = fields.Numeric("Settl. price",digits='unit') fixed_qt = fields.Numeric("Fixed qt",digits='unit', readonly=True) fixed_qt_price = fields.Numeric("Fixed qt price",digits='unit', readonly=True) unfixed_qt = fields.Numeric("Unfixed qt",digits='unit', readonly=True) unfixed_qt_price = fields.Numeric("Unfixed qt price",digits='unit', readonly=True) eod_price = fields.Numeric("EOD price",digits='unit',readonly=True) last = fields.Boolean("Last") @classmethod def default_fixed_qt(cls): return Decimal(0) @classmethod def default_unfixed_qt(cls): return Decimal(0) @classmethod def default_fixed_qt_price(cls): return Decimal(0) @classmethod def default_unfixed_qt_price(cls): return Decimal(0) @classmethod def default_quantity(cls): return Decimal(0) @classmethod def default_settl_price(cls): return Decimal(0) @classmethod def default_eod_price(cls): return Decimal(0) @classmethod def default_price_component(cls): Component = Pool().get('pricing.component') context = Transaction().context domains = [] if context.get('default_line'): domains.append([('line', '=', context['default_line'])]) if context.get('default_sale_line'): domains.append([('sale_line', '=', context['default_sale_line'])]) for domain in domains: component = cls._get_single_manual_component( Component.search(domain, limit=2)) if component: return component.id @fields.depends('line', 'sale_line') def on_change_with_available_components(self, name=None): Component = Pool().get('pricing.component') line = getattr(self, 'line', None) if line: return Component.search([('line', '=', line)]) sale_line = getattr(self, 'sale_line', None) if sale_line: return Component.search([('sale_line', '=', sale_line)]) return [] @classmethod def set_available_components(cls, records, name, value): pass @staticmethod def _get_single_manual_component(components): manual_components = [ component for component in components if not getattr(component, 'auto', False)] if len(manual_components) == 1 and len(components) == 1: return manual_components[0] def _default_single_manual_component(self): if self.price_component: return component = self._get_single_manual_component( self.on_change_with_available_components()) if component: self.price_component = component @fields.depends('line', 'sale_line', 'price_component') def on_change_line(self): self._default_single_manual_component() @fields.depends('line', 'sale_line', 'price_component') def on_change_sale_line(self): self._default_single_manual_component() @staticmethod def _weighted_average_price(fixed_qt, fixed_price, unfixed_qt, unfixed_price): fixed_qt = Decimal(str(fixed_qt or 0)) fixed_price = Decimal(str(fixed_price or 0)) unfixed_qt = Decimal(str(unfixed_qt or 0)) unfixed_price = Decimal(str(unfixed_price or 0)) total_qty = fixed_qt + unfixed_qt if total_qty == 0: return Decimal(0) return round( ((fixed_qt * fixed_price) + (unfixed_qt * unfixed_price)) / total_qty, 4, ) def compute_eod_price(self): if getattr(self, 'sale_line', None) and hasattr(self, 'get_eod_price_sale'): return self.get_eod_price_sale() if getattr(self, 'line', None) and hasattr(self, 'get_eod_price_purchase'): return self.get_eod_price_purchase() return self._weighted_average_price( self.fixed_qt, self.fixed_qt_price, self.unfixed_qt, self.unfixed_qt_price, ) @fields.depends('fixed_qt', 'fixed_qt_price', 'unfixed_qt', 'unfixed_qt_price') def on_change_fixed_qt(self): self.eod_price = self.compute_eod_price() @fields.depends('fixed_qt', 'fixed_qt_price', 'unfixed_qt', 'unfixed_qt_price') def on_change_fixed_qt_price(self): self.eod_price = self.compute_eod_price() @fields.depends('fixed_qt', 'fixed_qt_price', 'unfixed_qt', 'unfixed_qt_price') def on_change_unfixed_qt(self): self.eod_price = self.compute_eod_price() @fields.depends('fixed_qt', 'fixed_qt_price', 'unfixed_qt', 'unfixed_qt_price') def on_change_unfixed_qt_price(self): self.eod_price = self.compute_eod_price() @classmethod def create(cls, vlist): records = super(Pricing, cls).create(vlist) cls._sync_manual_values(records) cls._sync_manual_last(records) cls._sync_eod_price(records) return records @classmethod def write(cls, *args): super(Pricing, cls).write(*args) if (Transaction().context.get('skip_pricing_eod_sync') or Transaction().context.get('skip_pricing_last_sync')): return records = [] actions = iter(args) for record_set, values in zip(actions, actions): if values: records.extend(record_set) cls._sync_manual_values(records) cls._sync_manual_last(records) cls._sync_eod_price(records) @staticmethod def _component_matches_owner(record): def record_id(value): return getattr(value, 'id', value) component = getattr(record, 'price_component', None) if not component: return True line = getattr(record, 'line', None) if line: return record_id(getattr(component, 'line', None)) == record_id(line) sale_line = getattr(record, 'sale_line', None) if sale_line: return ( record_id(getattr(component, 'sale_line', None)) == record_id(sale_line)) return True @classmethod def validate(cls, records): super(Pricing, cls).validate(records) for record in records: if not cls._component_matches_owner(record): raise UserError( "Pricing component must belong to the related line.") @classmethod def _sync_eod_price(cls, records): if not records: return with Transaction().set_context(skip_pricing_eod_sync=True): for record in records: eod_price = record.compute_eod_price() if Decimal(str(record.eod_price or 0)) == Decimal(str(eod_price or 0)): continue super(Pricing, cls).write([record], { 'eod_price': eod_price, }) @classmethod def update_daily_pricing(cls): PurchaseLine = Pool().get('purchase.line') SaleLine = Pool().get('sale.line') purchase_lines = PurchaseLine.search([]) sale_lines = SaleLine.search([]) cls._update_line_pricing_prices(PurchaseLine, purchase_lines) cls._update_line_pricing_prices(SaleLine, sale_lines) logger.info( "Updated pricing for %s purchase line(s) and %s sale line(s)", len(purchase_lines), len(sale_lines)) @classmethod def _update_line_pricing_prices(cls, Line, lines): if not lines: return for line in lines: line.check_pricing() line._recompute_trade_price_fields() Line.save(lines) @classmethod def _is_manual_pricing_record(cls, record): component = getattr(record, 'price_component', None) if component is None: return True return not bool(getattr(component, 'auto', False)) @classmethod def _get_pricing_group_domain(cls, record): component = getattr(record, 'price_component', None) if getattr(record, 'sale_line', None): return [ ('sale_line', '=', record.sale_line.id), ('price_component', '=', component.id if getattr(component, 'id', None) else None), ] if getattr(record, 'line', None): return [ ('line', '=', record.line.id), ('price_component', '=', component.id if getattr(component, 'id', None) else None), ] return None @classmethod def _get_base_quantity(cls, record): owner = getattr(record, 'sale_line', None) or getattr(record, 'line', None) if not owner: return Decimal(0) if hasattr(owner, '_get_pricing_base_quantity'): return Decimal(str(owner._get_pricing_base_quantity() or 0)) quantity = getattr(owner, 'quantity_theorical', None) if quantity is None: quantity = getattr(owner, 'quantity', None) return Decimal(str(quantity or 0)) @classmethod def _get_manual_unfixed_price(cls, pricing, settl_price): component = getattr(pricing, 'price_component', None) if (not component or getattr(component, 'price_source_type', None) != 'curve' or not getattr(component, 'price_index', None)): return settl_price owner = getattr(pricing, 'sale_line', None) or getattr(pricing, 'line', None) if not owner: return settl_price document = getattr(owner, 'sale', None) or getattr(owner, 'purchase', None) currency = getattr(document, 'currency', None) unit = getattr(owner, 'unit', None) if not currency or not unit: return settl_price return round(Decimal(str(component.get_price( pricing.pricing_date, unit, currency, True) or 0)), 4) @classmethod def _sync_manual_values(cls, records): if (not records or Transaction().context.get('skip_pricing_manual_sync')): return domains = [] seen = set() for record in records: if not cls._is_manual_pricing_record(record): continue domain = cls._get_pricing_group_domain(record) if not domain: continue key = tuple(domain) if key in seen: continue seen.add(key) domains.append(domain) if not domains: return with Transaction().set_context( skip_pricing_manual_sync=True, skip_pricing_last_sync=True, skip_pricing_eod_sync=True): for domain in domains: pricings = cls.search( domain, order=[('pricing_date', 'ASC'), ('id', 'ASC')]) if not pricings: continue base_quantity = cls._get_base_quantity(pricings[0]) cumul_qt = Decimal(0) cumul_qt_price = Decimal(0) total = len(pricings) for index, pricing in enumerate(pricings): quantity = Decimal(str(pricing.quantity or 0)) settl_price = Decimal(str(pricing.settl_price or 0)) cumul_qt += quantity cumul_qt_price += quantity * settl_price fixed_qt = cumul_qt if fixed_qt > 0: fixed_qt_price = round(cumul_qt_price / fixed_qt, 4) else: fixed_qt_price = Decimal(0) unfixed_qt = base_quantity - fixed_qt if unfixed_qt < Decimal('0.001'): unfixed_qt = Decimal(0) fixed_qt = base_quantity values = { 'fixed_qt': fixed_qt, 'fixed_qt_price': fixed_qt_price, 'unfixed_qt': unfixed_qt, 'unfixed_qt_price': cls._get_manual_unfixed_price( pricing, settl_price), 'last': index == (total - 1), } eod_price = cls._weighted_average_price( values['fixed_qt'], values['fixed_qt_price'], values['unfixed_qt'], values['unfixed_qt_price'], ) values['eod_price'] = eod_price super(Pricing, cls).write([pricing], values) @classmethod def _get_manual_last_group_domain(cls, record): return cls._get_pricing_group_domain(record) @classmethod def _sync_manual_last(cls, records): if not records: return domains = [] seen = set() for record in records: domain = cls._get_manual_last_group_domain(record) if not domain: continue key = tuple(domain) if key in seen: continue seen.add(key) domains.append(domain) if not domains: return with Transaction().set_context( skip_pricing_last_sync=True, skip_pricing_eod_sync=True): for domain in domains: pricings = cls.search( domain, order=[('pricing_date', 'ASC'), ('id', 'ASC')]) if not pricings: continue last_pricing = pricings[-1] for pricing in pricings[:-1]: if pricing.last: super(Pricing, cls).write([pricing], {'last': False}) if not last_pricing.last: super(Pricing, cls).write([last_pricing], {'last': True}) def get_fixed_price(self): price = Decimal(0) Pricing = Pool().get('pricing.pricing') domain = self._get_pricing_group_domain(self) if not domain: return price pricings = Pricing.search(domain, order=[('pricing_date', 'ASC'), ('id', 'ASC')]) if pricings: cumul_qt = Decimal(0) cumul_qt_price = Decimal(0) for pr in pricings: quantity = Decimal(str(pr.quantity or 0)) settl_price = Decimal(str(pr.settl_price or 0)) cumul_qt += quantity cumul_qt_price += quantity * settl_price if pr.id == self.id: break if cumul_qt > 0: price = cumul_qt_price / cumul_qt return round(price,4) class Trigger(ModelSQL,ModelView): "Period rules" __name__ = "pricing.trigger" component = fields.Many2One('pricing.component',"Component", ondelete='CASCADE') pricing_period = fields.Many2One('pricing.period',"Pricing period") from_p = fields.Date("From", states={ 'readonly': Eval('pricing_period') != None, }) to_p = fields.Date("To", states={ 'readonly': Eval('pricing_period') != None, }) average = fields.Boolean("Avg") last = fields.Boolean("AL", help="Absolute Last") relative_last = fields.Boolean("RL", help="Relative Last") application_period = fields.Many2One('pricing.period',"Application period") from_a = fields.Date("From", states={ 'readonly': Eval('application_period') != None, }) to_a = fields.Date("To", states={ 'readonly': Eval('application_period') != None, }) @classmethod def default_average(cls): return True @fields.depends('pricing_period', 'application_period') def on_change_with_application_period(self): if self.application_period: return self.application_period if self.pricing_period: return self.pricing_period @fields.depends('from_p', 'from_a', 'application_period') def on_change_from_p(self): if not self.application_period and not self.from_a: self.from_a = self.from_p @fields.depends('to_p', 'to_a', 'application_period') def on_change_to_p(self): if not self.application_period and not self.to_a: self.to_a = self.to_p def getDateWithEstTrigger(self, period, cal=None): PP = Pool().get('pricing.period') if period == 1: pp = PP(self.pricing_period) else: pp = PP(self.application_period) CO = Pool().get('pricing.component') co = CO(self.component) if co.line: d = co.getEstimatedTriggerPurchase(pp.trigger) else: d = co.getEstimatedTriggerSale(pp.trigger) date_from,date_to,dates = pp.getDates(d, cal) return date_from,date_to,d,pp.include,dates def getApplicationListDates(self, cal): ld = [] dates = [] if self.application_period: date_from, date_to, d, include,dates = self.getDateWithEstTrigger(2, cal) else: date_from = self.from_a or self.from_p date_to = self.to_a or self.to_p d = None include = False ld, lprice = self.getListDates(date_from,date_to,d,include,cal,2,dates) return ld, lprice def getPricingListDates(self,cal): ld = [] dates = [] if self.pricing_period: date_from, date_to, d, include,dates = self.getDateWithEstTrigger(1, cal) else: date_from = self.from_p#datetime.datetime(self.from_p.year, self.from_p.month, self.from_p.day) date_to = self.to_p#datetime.datetime(self.to_p.year, self.to_p.month, self.to_p.day) d = None include = False ld, lprice = self.getListDates(date_from,date_to,d,include,cal,1,dates) return ld, lprice def getListDates(self,df,dt,t,i,cal,pricing,dates): l = [] lprice = [] def append_date(date): if cal: if not cal.IsQuote(date): return l.append(date) if pricing == 1: lprice.append(self.getprice(date)) if cal: CAL = Pool().get('price.calendar') cal = CAL(cal) if dates: for d in dates: append_date(d) return l, lprice if df and dt: current_date = datetime.datetime(df.year,df.month,df.day) dt = datetime.datetime(dt.year,dt.month,dt.day) while current_date <= dt: if i or (not i and current_date != t): append_date(current_date) current_date += datetime.timedelta(days=1) return l, lprice def getprice(self,current_date): PC = Pool().get('pricing.component') pc = PC(self.component) val = {} val['date'] = current_date line = pc.line if pc.line else pc.sale_line val['price'] = pc.get_price( current_date, line.unit, line.currency, self.last, self.relative_last) val['avg'] = val['price'] val['avg_minus_1'] = val['price'] val['isAvg'] = self.average return val class Period(ModelSQL,ModelView): "Period" __name__ = 'pricing.period' name = fields.Char("Name") trigger = fields.Selection(TRIGGERS, 'Trigger') include = fields.Boolean("Inc.") startday = fields.Selection(DAYTYPES,"Start day") nbds = fields.Integer("Nb") endday = fields.Selection(DAYTYPES,"End day") nbde = fields.Integer("Nb") nbms = fields.Integer("Starting month") nbme = fields.Integer("Ending month") every = fields.Selection(DAYS,"Every") nb_quotation_before = fields.Integer("Nb quotes before") nb_quotation_after = fields.Integer("Nb quotes after") nb_quotation = fields.Integer("Nb quotation") @classmethod def default_nbds(cls): return 0 @classmethod def default_nbde(cls): return 0 @classmethod def default_nbms(cls): return 0 @classmethod def default_nbme(cls): return 0 @classmethod def default_nb_quotation_before(cls): return 0 @classmethod def default_nb_quotation_after(cls): return 0 @staticmethod def _calendar_is_quote(cal, date): if not cal: return True if not hasattr(cal, 'IsQuote'): Calendar = Pool().get('price.calendar') cal = Calendar(cal) result = cal.IsQuote(date) return True if result is None else bool(result) @staticmethod def _as_datetime(date): return datetime.datetime(date.year, date.month, date.day) def _available_quotation_dates(self, trigger_date, count, step, cal=None): dates = [] current = trigger_date + datetime.timedelta(days=step) while len(dates) < count: date = self._as_datetime(current) if self._calendar_is_quote(cal, date): dates.append(date) current += datetime.timedelta(days=step) if step < 0: dates.reverse() return dates def _quotation_counts(self): before = self.nb_quotation_before or 0 after = self.nb_quotation_after or 0 if before or after or not self.nb_quotation: return max(before, 0), max(after, 0) if self.every == 'before_after': count = abs(self.nb_quotation) return count, count if self.nb_quotation < 0: return abs(self.nb_quotation), 0 return 0, self.nb_quotation @staticmethod def _weekday_dates(trigger_date, weekday_target, count, step): dates = [] current = trigger_date while len(dates) < count: current += datetime.timedelta(days=step) if current.weekday() == weekday_target: dates.append(datetime.datetime( current.year, current.month, current.day)) if step < 0: dates.reverse() return dates def getDates(self,t, cal=None): date_from = None date_to = None dates = [] if t: before_count, after_count = self._quotation_counts() if self.every: if t: j = self.every if j == 'before_after': dates.extend( self._available_quotation_dates( t, before_count, -1, cal)) trigger_date = self._as_datetime(t) if (self.include and self._calendar_is_quote( cal, trigger_date)): dates.append(trigger_date) dates.extend( self._available_quotation_dates( t, after_count, 1, cal)) return date_from, date_to, dates if j not in WEEKDAY_MAP: raise ValueError(f"Invalid day : '{j}'") weekday_target = WEEKDAY_MAP[j] if self.trigger == 'delmonth': first_day = t.replace(day=1) days_to_add = (weekday_target - first_day.weekday()) % 7 current = first_day + datetime.timedelta(days=days_to_add) while current.month == t.month: dates.append(datetime.datetime(current.year, current.month, current.day)) current += datetime.timedelta(days=7) elif before_count or after_count: dates.extend(self._weekday_dates( t, weekday_target, before_count, -1)) trigger_date = self._as_datetime(t) if (self.include and trigger_date.weekday() == weekday_target): dates.append(trigger_date) dates.extend(self._weekday_dates( t, weekday_target, after_count, 1)) elif before_count or after_count: dates.extend( self._available_quotation_dates( t, before_count, -1, cal)) trigger_date = self._as_datetime(t) if (self.include and self._calendar_is_quote(cal, trigger_date)): dates.append(trigger_date) dates.extend( self._available_quotation_dates( t, after_count, 1, cal)) else: if self.startday == 'before': date_from = t - datetime.timedelta(days=(self.nbds if self.nbds else 0)) elif self.startday == 'after': date_from = t + datetime.timedelta(days=(self.nbds if self.nbds else 0)) elif self.startday == 'first': date_from = datetime.datetime(t.year, t.month % 12 + (self.nbms if self.nbms else 0), 1) elif self.startday == 'last': date_from = datetime.datetime(t.year, t.month % 12 + 1, 1) - datetime.timedelta(days=1) elif self.startday == 'xth': date_from = datetime.datetime(t.year, t.month % 12, (self.nbds if self.nbds else 1)) else: date_from = datetime.datetime(t.year, t.month, t.day) if self.endday == 'before': date_to = t - datetime.timedelta(days=(self.nbde if self.nbde else 0)) elif self.endday == 'after': date_to = t + datetime.timedelta(days=(self.nbde if self.nbde else 0)) elif self.endday == 'first': date_to = datetime.datetime(t.year, t.month % 12 + (self.nbme if self.nbme else 0), 1) elif self.endday == 'last': date_to = datetime.datetime(t.year, t.month % 12 + 1, 1) - datetime.timedelta(days=1) elif self.endday == 'xth': date_to = datetime.datetime(t.year, t.month % 12, (self.nbds if self.nbds else 1)) else: date_to = date_from return date_from, date_to, dates