1091 lines
38 KiB
Python
Executable File
1091 lines
38 KiB
Python
Executable File
# 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.exceptions import UserError
|
|
from trytond.pool import Pool, PoolMeta
|
|
from trytond.pyson import Bool, Eval, Id
|
|
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, 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)
|
|
|
|
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'),
|
|
]
|
|
|
|
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"),
|
|
('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',
|
|
'openprice': 'open_price',
|
|
'pricevalue': 'price_value',
|
|
}
|
|
|
|
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)])
|
|
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):
|
|
values = {
|
|
'price_index': price_index,
|
|
'price_desc': 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')]),
|
|
]
|
|
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'(?<!\d)(20\d{2})[-_/\. ](0[1-9]|1[0-2])(?!\d)',
|
|
price_index)
|
|
if not match:
|
|
return None
|
|
|
|
year = int(match.group(1))
|
|
month = int(match.group(2))
|
|
month_name = cls._period_month_name(year, month)
|
|
periods = cls._period_model().search(
|
|
[('month_name', '=', month_name)], limit=1)
|
|
if periods:
|
|
return periods[0]
|
|
|
|
beg_date = datetime.date(year, month, 1)
|
|
end_date = datetime.date(
|
|
year, month, calendar.monthrange(year, month)[1])
|
|
period, = cls._period_model().create([{
|
|
'month_name': month_name,
|
|
'description': month_name,
|
|
'beg_date': beg_date,
|
|
'end_date': end_date,
|
|
'is_cotation': True,
|
|
}])
|
|
return period
|
|
|
|
@staticmethod
|
|
def _period_month_name(year, month):
|
|
return '%s%s' % (
|
|
calendar.month_abbr[month].upper(), str(year)[-2:])
|
|
|
|
@staticmethod
|
|
def _period_model():
|
|
return Pool().get('product.month')
|
|
|
|
@staticmethod
|
|
def _first_record(model_name, domain):
|
|
records = Pool().get(model_name).search(domain, limit=1)
|
|
return records[0] if records else None
|
|
|
|
@classmethod
|
|
def _price_value_values(cls, price, row, price_date):
|
|
return {
|
|
'price': price.id,
|
|
'price_date': price_date,
|
|
'high_price': cls._as_float(row.get('high_price')),
|
|
'low_price': cls._as_float(row.get('low_price')),
|
|
'open_price': cls._as_float(row.get('open_price')),
|
|
'price_value': cls._as_float(row.get('price_value')),
|
|
}
|
|
|
|
@staticmethod
|
|
def _result_line(row_number, price_index, price_date, detail):
|
|
return {
|
|
'row': row_number,
|
|
'price_index': price_index,
|
|
'price_date': price_date,
|
|
'detail': detail,
|
|
}
|
|
|
|
@classmethod
|
|
def _format_result(cls, stats):
|
|
lines = [
|
|
'Import completed.',
|
|
f"Created price indexes: {len(stats['created_indexes'])}",
|
|
f"Imported prices: {len(stats['imported'])}",
|
|
f"Updated prices: {len(stats['updated'])}",
|
|
f"Skipped rows: {len(stats['skipped'])}",
|
|
f"Errors: {len(stats['errors'])}",
|
|
]
|
|
|
|
sections = [
|
|
('Created price indexes', stats['created_indexes']),
|
|
('Successfully imported prices', stats['imported']),
|
|
('Updated existing prices', stats['updated']),
|
|
('Skipped records', stats['skipped']),
|
|
('Errors', stats['errors']),
|
|
]
|
|
for title, items in sections:
|
|
lines.append('')
|
|
lines.append('%s:' % title)
|
|
if not items:
|
|
lines.append('- None')
|
|
continue
|
|
for item in items:
|
|
lines.append('- %s' % cls._format_result_line(item))
|
|
return '\n'.join(lines)
|
|
|
|
@staticmethod
|
|
def _format_result_line(item):
|
|
label = item['price_index'] or '<empty price_index>'
|
|
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', '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)
|
|
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_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,
|
|
'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 _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 = fields.Date("Valuation Date", required=True)
|
|
use_last_price = fields.Boolean("Use Last Available Price")
|
|
calendar = fields.Many2One(
|
|
'price.calendar', "Calendar"
|
|
)
|
|
|
|
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
|
|
|
|
def get_mtm(self,line,qty):
|
|
pool = Pool()
|
|
Currency = pool.get('currency.currency')
|
|
total = Decimal(0)
|
|
|
|
scenario = self.scenario
|
|
dt = scenario.valuation_date
|
|
|
|
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,
|
|
last=scenario.use_last_price
|
|
)
|
|
)
|
|
|
|
elif comp.price_source_type == 'matrix' and comp.price_matrix:
|
|
value = self._get_matrix_price(comp, line, dt)
|
|
|
|
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()
|
|
Snapshot.create([{
|
|
'strategy': strat.id,
|
|
'valuation_date': strat.scenario.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
|
|
return None
|
|
|
|
@fields.depends('price_index','price_matrix')
|
|
def on_change_with_currency(self):
|
|
return self.get_cur()
|
|
|
|
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"
|
|
)
|
|
|
|
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'),
|
|
# ('manual', 'Manual'),
|
|
], "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")
|
|
price_matrix = fields.Many2One('price.matrix', "Price Matrix")
|
|
currency = fields.Function(fields.Many2One('currency.currency',"Curr."),'get_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
|
|
else:
|
|
return '[' + self.fix_type.name + '] '
|
|
|
|
def get_cur(self,name):
|
|
if self.price_index:
|
|
PI = Pool().get('price.price')
|
|
pi = PI(self.price_index)
|
|
return pi.price_currency
|
|
|
|
def get_nbdays(self, name):
|
|
days = 0
|
|
if self.triggers:
|
|
for t in self.triggers:
|
|
l,l2 = t.getApplicationListDates(self.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('line.price_components'))], ondelete='CASCADE')
|
|
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)
|
|
|
|
def get_fixed_price(self):
|
|
price = Decimal(0)
|
|
Pricing = Pool().get('pricing.pricing')
|
|
pricings = Pricing.search(['price_component','=',self.price_component.id],order=[('pricing_date', 'ASC')])
|
|
if pricings:
|
|
cumul_qt = Decimal(0)
|
|
cumul_qt_price = Decimal(0)
|
|
for pr in pricings:
|
|
cumul_qt += pr.quantity
|
|
cumul_qt_price += pr.quantity * pr.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("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,
|
|
})
|
|
|
|
@fields.depends('pricing_period')
|
|
def on_change_with_application_period(self):
|
|
if not self.application_period and self.pricing_period:
|
|
return self.pricing_period
|
|
|
|
def getDateWithEstTrigger(self, period):
|
|
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)
|
|
return date_from,date_to,d,pp.include,dates
|
|
|
|
def getApplicationListDates(self, cal):
|
|
ld = []
|
|
if self.application_period:
|
|
date_from, date_to, d, include,dates = self.getDateWithEstTrigger(2)
|
|
else:
|
|
date_from = self.from_a
|
|
date_to = self.to_a
|
|
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 = []
|
|
if self.pricing_period:
|
|
date_from, date_to, d, include,dates = self.getDateWithEstTrigger(1)
|
|
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 = []
|
|
CAL = Pool().get('price.calendar')
|
|
if cal:
|
|
cal = CAL(cal)
|
|
if dates:
|
|
for d in dates:
|
|
if cal.IsQuote(d):
|
|
l.append(d)
|
|
if pricing == 1:
|
|
lprice.append(self.getprice(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):
|
|
if cal:
|
|
if cal.IsQuote(current_date):
|
|
l.append(current_date)
|
|
if pricing == 1:
|
|
lprice.append(self.getprice(current_date))
|
|
else:
|
|
l.append(current_date)
|
|
if pricing == 1:
|
|
lprice.append(self.getprice(current_date))
|
|
current_date += datetime.timedelta(days=1)
|
|
return l, lprice
|
|
|
|
def getprice(self,current_date):
|
|
PI = Pool().get('price.price')
|
|
PC = Pool().get('pricing.component')
|
|
pc = PC(self.component)
|
|
pi = PI(pc.price_index)
|
|
val = {}
|
|
val['date'] = current_date
|
|
val['price'] = pi.get_price(current_date,pc.line.unit if pc.line else pc.sale_line.unit,pc.line.currency if pc.line else pc.sale_line.currency,self.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 = 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
|
|
|
|
def getDates(self,t):
|
|
date_from = None
|
|
date_to = None
|
|
dates = []
|
|
if t:
|
|
if self.every:
|
|
if t:
|
|
j = self.every
|
|
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 self.nb_quotation > 0:
|
|
days_to_add = (weekday_target - t.weekday()) % 7
|
|
current = t + datetime.timedelta(days=days_to_add)
|
|
while len(dates) < self.nb_quotation:
|
|
dates.append(datetime.datetime(current.year, current.month, current.day))
|
|
current += datetime.timedelta(days=7)
|
|
elif self.nb_quotation < 0:
|
|
days_to_sub = (t.weekday() - weekday_target) % 7
|
|
current = t - datetime.timedelta(days=days_to_sub)
|
|
while len(dates) < -self.nb_quotation:
|
|
dates.append(datetime.datetime(current.year, current.month, current.day))
|
|
current -= datetime.timedelta(days=7)
|
|
|
|
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
|