| 1 | """Materialize price ratios without aging the earnings/book basis every day.""" |
| 2 | from datetime import timedelta |
| 3 | from decimal import Decimal, InvalidOperation |
| 4 | from app.models import ProvenancedValue |
| 5 | |
| 6 | def materialize_valuation(records, prices, *, now): |
| 7 | usable=[p for p in prices if p.observed_at<=now and p.retrieved_at<=now and p.price>0] |
| 8 | if not usable: return {} |
| 9 | stamp=max(p.observed_at for p in usable) |
| 10 | latest=[p for p in usable if p.observed_at==stamp] |
| 11 | if max(p.price for p in latest)!=min(p.price for p in latest): return {} |
| 12 | price=min(latest,key=lambda p:(p.provider,p.source_url)) |
| 13 | output={} |
| 14 | for source_name,target in (('trailingEps','trailingPE'),('bookValue','priceToBook')): |
| 15 | bases=[] |
| 16 | for record in records: |
| 17 | if record.instrument_id != price.instrument_id or not record.currency or record.currency != price.currency: |
| 18 | continue |
| 19 | value=record.snapshot.facts.get(source_name) |
| 20 | if value is None: continue |
| 21 | anchor=value.as_of_date or value.published_at or value.retrieved_at |
| 22 | if value.retrieved_at>now or anchor>now or now-anchor>timedelta(days=120): continue |
| 23 | try: number=Decimal(str(value.value)) |
| 24 | except InvalidOperation: continue |
| 25 | if number.is_finite() and number>0: bases.append((anchor,value.source_url,number,value)) |
| 26 | if not bases: continue |
| 27 | _,_,number,basis=max(bases,key=lambda b:(b[0],b[1])) |
| 28 | output[target]=ProvenancedValue(value=price.price/number,as_of_date=stamp,retrieved_at=max(price.retrieved_at,basis.retrieved_at), |
| 29 | source_url=price.source_url,source_name='Persisted price / valid '+source_name, |
| 30 | source_type='DERIVED_PERSISTED_VALUATION',confidence=basis.confidence, |
| 31 | calculation_basis=f'price={price.source_url}; basis={basis.source_url}; basisAsOf={basis.as_of_date or basis.retrieved_at}') |
| 32 | return output |