junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1
Model/etl/customers/goldman/property_ownership.py
Khalim Conn-Kowlessar 39bc6c53b8 sfr investiation in progress
2024-08-01 12:21:10 +01:00

1241 lines
53 KiB
Python
Raw Permalink Blame History

import re
import pandas as pd
from tqdm import tqdm
import Levenshtein
from backend.SearchEpc import SearchEpc
from utils.s3 import read_dataframe_from_s3_parquet
# Average value of a property in the midlands in 2024 was £238,000. Since these are EPC F & G properties, we assume
# £207,000 since they trade at a discount. This is based on the rightmove study where moving from an EPC F/G -> C has a
# +15% impact on valuation and D -> C has a +3% impact on valuation.
# The mode EPC rating is D, so we associate the £238k valuation with an EPC D property
# Therefore value_of_F * 1.15 = value_of_D * 1.03
# Therefore value_of_F = value_of_D * 1.03/1.15 = 238k * (1.03/1.15) = 213165
PROPERTY_VALUE_ESTIMATE = 200_000
# UPRNs of properties we need
MANUAL_EXCLUSIONS = []
def aggregate_matches(matching_lookup, company_ownership, properties):
df = matching_lookup.merge(
company_ownership, how="left", on="Title Number"
).merge(
properties[["UPRN", "LOCAL_AUTHORITY_LABEL"]], how="left", on="UPRN"
)
counts = (
df.groupby(["Company Registration No. (1)", "LOCAL_AUTHORITY_LABEL"])["UPRN"]
.count()
.reset_index(name="number_of_properties")
)
counts = counts.sort_values("number_of_properties", ascending=False)
pivot_counts = counts.pivot_table(
index=["Company Registration No. (1)"], # Rows: companies and proprietors
columns="LOCAL_AUTHORITY_LABEL", # Columns: each local authority
values="number_of_properties", # The counts of properties
fill_value=0 # Fill missing values with 0 (where there are no properties owned)
).reset_index()
total_counts = (
df.groupby(["Company Registration No. (1)"])["UPRN"]
.count()
.reset_index(name="total_number_of_properties")
)
# We have cases where the same company registration number results in the same company name, so we produce a best
# name per company registration number
best_names = (
df.groupby(["Company Registration No. (1)"])["Proprietor Name (1)"]
.first()
.reset_index()
)
total_counts = best_names.merge(
total_counts, how="left", on=["Company Registration No. (1)"]
)
pivot_counts = pivot_counts.merge(
total_counts, how="left", on=["Company Registration No. (1)"]
)
pivot_counts = pivot_counts.sort_values("total_number_of_properties", ascending=False)
pivot_counts["approx_value"] = PROPERTY_VALUE_ESTIMATE * pivot_counts["total_number_of_properties"]
pivot_counts["cumulative_value"] = pivot_counts["approx_value"].cumsum()
return pivot_counts
def find_f_g_properties(paths):
data = []
for path in tqdm(paths):
epc_data = pd.read_csv(path, low_memory=False)
epc_data = epc_data[~pd.isnull(epc_data["UPRN"])]
epc_data["UPRN"] = epc_data["UPRN"].astype(int).astype(str)
if pd.isnull(pd.to_datetime(epc_data["LODGEMENT_DATETIME"], errors="coerce")).sum():
raise Exception("wtf")
# Get the newest EPC for each UPRN. We use LODGEMENT_DATE as a proxy for this
epc_data["LODGEMENT_DATETIME"] = pd.to_datetime(epc_data["LODGEMENT_DATETIME"], errors="coerce")
epc_data = epc_data.sort_values(
["LODGEMENT_DATE", "LODGEMENT_DATETIME"], ascending=False
).drop_duplicates("UPRN")
# Get G & F properties
epc_data = epc_data[epc_data["CURRENT_ENERGY_RATING"].isin(["G", "F"])]
data.append(epc_data)
data = pd.concat(data)
# Save as an excel
data.to_excel("EPC F & G Properties - V2.xlsx", index=False)
def remove_text_in_brackets(address: str) -> str:
"""
Removes any text within parentheses, including the parentheses themselves.
Parameters:
- address (str): The address string to clean.
Returns:
- str: The cleaned address with text in parentheses removed.
"""
# Regex to find and remove content in parentheses
cleaned_address = re.sub(r'\s*\([^)]*\)', '', address)
return cleaned_address
def extract_numeric_part(house_number: str) -> str:
"""
Extracts only the numeric part from a house number that may contain letters.
Parameters:
- house_number (str): The house number string possibly containing letters.
Returns:
- str: The numeric part of the house number.
"""
# Use regular expression to replace all non-digit characters with nothing
numeric_part = re.sub(r'\D', '', house_number)
return numeric_part
def levenstein_match(matching_string, df, address_col):
match_to = df[address_col].tolist()
# Strip out punctuation and spaces
match_to = [re.sub(r'[^\w\s]', '', x) for x in match_to]
match_to = [x.replace(" ", "") for x in match_to]
# Perform matching between full key and match_to
distances = [Levenshtein.distance(matching_string, s) for s in match_to]
best_match_index = distances.index(min(distances))
# We might want to consider a threshold for the distance, however for the momeny,
# we don't consider this for the moment
df = df.iloc[best_match_index:best_match_index + 1]
return df
def extract_range_from_house_number(house_number_range: str):
"""
Detects if the house number includes a numeric range (formatted as 'x-y') and extracts all values within this range.
Non-numeric strings containing hyphens are ignored.
Parameters:
- house_number_range (str): The house number string that might contain a range.
Returns:
- list of str: A list of all numbers within the range if it is a range; otherwise, returns None.
"""
if not house_number_range:
return None
if '-' in house_number_range:
parts = house_number_range.split('-')
if len(parts) == 2 and parts[0].isdigit() and parts[1].isdigit():
# Both parts are numeric, so it's a valid range
start, end = map(int, parts) # Convert parts to integers
return [str(x) for x in range(start, end + 1)]
else:
# Not a valid numeric range
return None
else:
# No hyphen present or not a range
return None
def is_in_range(row, house_no):
""" Check if the house number is within the range provided in the row. """
if row and any(house_no == num for num in row):
return True
return False
def remove_duplicate_matches(matching_lookup, properties, company_ownership):
duplicated_titles = matching_lookup[matching_lookup["Title Number"].duplicated()]["Title Number"].unique()
to_drop = []
for dupe_title in duplicated_titles:
dupe_data = matching_lookup[matching_lookup["Title Number"] == dupe_title].copy()
matched_addresses = dupe_data.merge(
properties[["UPRN", "ADDRESS"]].rename(columns={"ADDRESS": "epc_address"}),
how="left", on="UPRN"
).merge(
company_ownership[["Title Number", "Property Address"]],
how="left", on="Title Number"
)
# We perform levenstein to get the best match
best_match = levenstein_match(
matching_string=matched_addresses["Property Address"].values[0],
df=matched_addresses,
address_col="epc_address"
)
matches_to_drop = matched_addresses[
~matched_addresses["UPRN"].isin(best_match["UPRN"].values)
]
to_drop.append(
matches_to_drop[["UPRN", "Title Number"]].copy()
)
to_drop = pd.concat(to_drop) if to_drop else pd.DataFrame()
if not to_drop.empty:
merged = pd.merge(matching_lookup, to_drop, on=['UPRN', 'Title Number'], how='left', indicator=True)
merged = merged[merged['_merge'] == 'left_only'].drop(columns=['_merge'])
return merged
return matching_lookup
def remove_duplicate_uprn_matches(matching_lookup, properties, company_ownership):
dupe_uprns = matching_lookup[matching_lookup["UPRN"].duplicated()]["UPRN"].unique().tolist()
to_drop = []
for dupe_uprn in dupe_uprns:
dupe_data = matching_lookup[matching_lookup["UPRN"] == dupe_uprn].copy()
matched_addresses = dupe_data.merge(
properties[["UPRN", "ADDRESS"]].rename(columns={"ADDRESS": "epc_address"}),
how="left", on="UPRN"
).merge(
company_ownership[["Title Number", "Property Address"]],
how="left", on="Title Number"
)
# We perform levenstein to get the best match
best_match = levenstein_match(
matching_string=matched_addresses["Property Address"].values[0],
df=matched_addresses,
address_col="epc_address"
)
matches_to_drop = matched_addresses[
~matched_addresses["Title Number"].isin(best_match["Title Number"].values)
]
to_drop.append(
matches_to_drop[["UPRN", "Title Number"]].copy()
)
to_drop = pd.concat(to_drop)
if not to_drop.empty:
merged = pd.merge(matching_lookup, to_drop, on=['UPRN', 'Title Number'], how='left', indicator=True)
merged = merged[merged['_merge'] == 'left_only'].drop(columns=['_merge'])
return merged
return matching_lookup
def filter_land_registry(properties):
column_names = [
"transaction_id",
"price",
"date_of_transfer",
"postcode",
"property_type",
"old_new",
"duration",
"paon",
"saon",
"street",
"locality",
"town_city",
"district",
"county",
"ppd_category_type",
"record_status",
]
land_registry = pd.read_csv("/Users/khalimconn-kowlessar/Downloads/pp-complete.csv", header=None)
land_registry.columns = column_names
land_registry = land_registry[
land_registry["postcode"].str.lower().isin(properties["POSTCODE"].str.lower().unique())
]
land_registry["date_of_transfer"] = pd.to_datetime(
land_registry["date_of_transfer"], format="%Y-%m-%d", errors="coerce"
)
# Take data from the last 5 years
land_registry = land_registry[
(land_registry["date_of_transfer"] >= "2019-01-01")
]
# Filter this
land_registry.to_csv(
"/Users/khalimconn-kowlessar/Downloads/land_registry_prices_paid_filtered.csv", index=False
)
def is_substring(x, match_string):
if pd.isnull(x):
return False
return x in match_string.lower()
def house_number_match(paon, house_number):
# Firstly try and convert to numberic
try:
paon_numeric = int(paon)
house_number_numeric = int(house_number)
return paon_numeric == house_number_numeric
except Exception as e: # noqa
# If we can't convert both to numeric, we do an equality
return paon == house_number
def check_equalities(lr_filtered):
all_paon_equal = all(lr_filtered["paon"] == lr_filtered["paon"].values[0])
if pd.isnull(lr_filtered["saon"].values[0]):
all_saon_equal = all(pd.isnull(lr_filtered["saon"]))
else:
all_saon_equal = all(lr_filtered["saon"] == lr_filtered["saon"].values[0])
all_street_equal = all(lr_filtered["street"] == lr_filtered["street"].values[0])
return all_paon_equal, all_saon_equal, all_street_equal
def app():
"""
This script is for scoping property ownership for EPC F & G rated properties in Birmingam, for Goldman Sachs
"""
# TODO: This property:
# https://epc.opendatacommunities.org/domestic/search?address=&postcode=&local-authority=&constituency
# =&uprn=100031179243&from-month=1&from-year=2008&to-month=12&to-year=2024
# is actually listed in two local authorities causing us to think it's an EPC F & G property, but it's
# it's actually EPC E. Need to handle this, probably by reading in all of the EPC data, concatenating
# together and performing a singular filter for most recent EPC by UPRN
# paths = [
# "local_data/all-domestic-certificates/domestic-E08000025-Birmingham/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E08000031-Wolverhampton/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E08000026-Coventry/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000016-Leicester/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000015-Derby/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000021-Stoke-on-Trent/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000018-Nottingham/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000154-Northampton/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000061-North-Northamptonshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000062-West-Northamptonshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000152-East-Northamptonshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000155-South-Northamptonshire/certificates.csv",
# #
# "local_data/all-domestic-certificates/domestic-E08000027-Dudley/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E08000029-Solihull/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000234-Bromsgrove/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E08000030-Walsall/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E08000028-Sandwell/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000019-Herefordshire-County-of/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000020-Telford-and-Wrekin/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000218-North-Warwickshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000222-Warwick/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000237-Worcester/certificates.csv",
# # East midlands
# "local_data/all-domestic-certificates/domestic-E07000035-Derbyshire-Dales/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000038-North-East-Derbyshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000039-South-Derbyshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000012-North-East-Lincolnshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000013-North-Lincolnshire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000138-Lincoln/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E07000134-North-West-Leicestershire/certificates.csv",
# "local_data/all-domestic-certificates/domestic-E06000017-Rutland/certificates.csv",
# ]
# paths = list(set(paths))
# find_f_g_properties(paths)
properties = pd.read_excel("EPC F & G Properties - V2.xlsx")
# filter_land_registry(properties)
company_ownership = pd.read_csv("/Users/khalimconn-kowlessar/Downloads/CCOD_FULL_2024_07.csv")
company_ownership["is_overseas"] = False
overseas_company_ownership = pd.read_csv("/Users/khalimconn-kowlessar/Downloads/OCOD_FULL_2024_07.csv")
overseas_company_ownership["is_overseas"] = True
company_ownership = pd.concat([company_ownership, overseas_company_ownership])
# FIlter on relevant postcodes
company_ownership = company_ownership[
company_ownership["Postcode"].str.lower().isin(properties["POSTCODE"].str.lower().unique())
]
# Now we filter properties the other way around
properties = properties[properties["POSTCODE"].str.lower().isin(company_ownership["Postcode"].str.lower().unique())]
# We end up with 7.4k entires on a postcode match, however we need to now do a direct address match
# Take just private rentals
properties = properties[
properties["TENURE"].isin(["rental (private)", "Rented (private)", "owner-occupied", "Owner-occupied"])
]
# We have some duplicated on UPRN
# Take the newest UPRN
properties = properties.sort_values("LODGEMENT_DATE", ascending=False).drop_duplicates("UPRN")
# Remove entries where the address begins with the term "land adjoining", or other records that don't reference the
# the property itself
starting_terms = [
"land adjoining", "land on the", "land to the rear of", "land and buildings on the",
"garage adjoining", "car park adjoining", "the land adjoining", "land and buildings adjoining",
"all royal mines"
]
for starting_term in starting_terms:
company_ownership = company_ownership[
~company_ownership["Property Address"].str.lower().str.startswith(starting_term)
]
# address = properties[properties["UPRN"] == 100030253055].squeeze()
freehold_matching_lookup = [] # 634
leasehold_matching_lookup = [] # 86
shared_leasehold_match = []
shared_freehold_match = []
for _, address in tqdm(properties.iterrows(), total=len(properties)):
match_type = "exact"
filtered = company_ownership[
company_ownership["Postcode"].str.lower() == address["POSTCODE"].lower()
].copy()
# Remove postcode and remove trailing commas
filtered["house_number"] = (
filtered["Property Address"]
.apply(remove_text_in_brackets)
.apply(SearchEpc.get_house_number)
.str.lower()
.str.replace(",", "")
)
house_no = SearchEpc.get_house_number(address["ADDRESS1"])
if house_no is not None:
house_no = house_no.replace(",", "")
if house_no is None:
# It's hard for us to get a reliable match
# filtered = filtered[filtered["Property Address"].str.contains(address["ADDRESS1"])]
# if filtered.shape[0] > 1:
# raise Exception("No valid - maybe we should do levenstein?")
continue
else:
if house_no not in filtered["house_number"].values:
# If this happens, we check house_number for a x-y range of addresses
filtered["house_number_range"] = filtered["house_number"].apply(extract_range_from_house_number)
# If we have found a house number range, we check if the house number is in the range and if not,
# we drop the row
filtered['is_in_range'] = filtered['house_number_range'].apply(lambda x: is_in_range(x, house_no))
if filtered['is_in_range'].any():
# If house_no is found in any range, keep only rows where it is in range
filtered = filtered[filtered['is_in_range']]
else:
# If house_no is not found in any range, filter out rows where 'house_number_range' is not None
filtered = filtered[filtered['house_number_range'].isnull()]
# Strip out letters from house_no and house_number
house_no = extract_numeric_part(house_no)
filtered["house_number"] = filtered["house_number"].astype(str).apply(extract_numeric_part)
match_type = "approximate"
filtered = filtered[filtered["house_number"] == house_no]
if filtered.empty:
continue
filtered_freehold = filtered[filtered["Tenure"] == "Freehold"]
filtered_leasehold = filtered[filtered["Tenure"] == "Leasehold"]
if filtered_freehold.shape[0] > 1:
matched = filtered_leasehold[["Title Number"]].copy()
matched.insert(0, "UPRN", address["UPRN"])
shared_freehold_match.append(matched)
elif not filtered_freehold.empty:
freehold_matching_lookup.append(
{
"UPRN": address["UPRN"],
"Title Number": filtered_freehold["Title Number"].values[0],
"match_type": match_type,
}
)
if filtered_leasehold.shape[0] > 1:
matched = filtered_leasehold[["Title Number"]].copy()
matched.insert(0, "UPRN", address["UPRN"])
shared_leasehold_match.append(matched)
elif not filtered_leasehold.empty:
leasehold_matching_lookup.append(
{
"UPRN": address["UPRN"],
"Title Number": filtered_leasehold["Title Number"].values[0],
"match_type": match_type,
}
)
freehold_matching_lookup = pd.DataFrame(freehold_matching_lookup)
leasehold_matching_lookup = pd.DataFrame(leasehold_matching_lookup)
# freehold_matching_lookup.to_excel("freehold_matching_lookup V2.xlsx")
# leasehold_matching_lookup.to_excel("leasehold_matching_lookup V2.xlsx")
# freehold_matching_lookup = pd.read_excel("freehold_matching_lookup V2.xlsx")
# leasehold_matching_lookup = pd.read_excel("leasehold_matching_lookup V2.xlsx")
# freehold_matching_lookup.shape
# (1537, 4)
# leasehold_matching_lookup.shape
# (390, 4)
# The approximate matches aren't very good
freehold_matching_lookup = freehold_matching_lookup[freehold_matching_lookup["match_type"] == "exact"]
leasehold_matching_lookup = leasehold_matching_lookup[leasehold_matching_lookup["match_type"] == "exact"]
# Combine
combined_matching_lookup = pd.concat([freehold_matching_lookup, leasehold_matching_lookup])
# Remove duplicates
combined_matching_lookup = remove_duplicate_matches(
matching_lookup=combined_matching_lookup, properties=properties, company_ownership=company_ownership
)
# We also have duplicates at a UPRN level
combined_matching_lookup = remove_duplicate_uprn_matches(combined_matching_lookup, properties, company_ownership)
matched_addresses = combined_matching_lookup.merge(
properties[
[
"UPRN",
"ADDRESS",
"ADDRESS1",
"CURRENT_ENERGY_EFFICIENCY",
"CURRENT_ENERGY_RATING",
"POSTCODE",
"LODGEMENT_DATE",
"TRANSACTION_TYPE"
]
].rename(
columns={
"ADDRESS": "epc_address",
"ADDRESS1": "epc_address1",
"POSTCODE": "epc_postcode"
}
),
how="left", on="UPRN"
).merge(
company_ownership[
[
"Title Number",
"Property Address",
"Postcode",
"Company Registration No. (1)",
"Proprietor Name (1)",
"Date Proprietor Added",
]
],
how="left", on="Title Number"
)
# Let's try and get the house number
matched_addresses["house_number"] = (
matched_addresses["epc_address"]
.apply(remove_text_in_brackets)
.apply(SearchEpc.get_house_number)
.str.lower()
.str.replace(",", "")
)
# Read in land registry
land_registry = pd.read_csv(
"/Users/khalimconn-kowlessar/Downloads/land_registry_prices_paid_filtered.csv",
)
# We now perform a match between the land registry data and the matched address, in an attempt to find
# out when these properties last sold. The land registry data has been pre filtered on the postcodes in this
# data, and for sales within the last 5 years, to ensure the file isn't too large.
land_registry["postcode"] = land_registry["postcode"].str.lower().str.strip()
land_registry["street"] = land_registry["street"].str.lower().str.strip()
land_registry["paon"] = land_registry["paon"].str.lower().str.strip()
land_registry["saon"] = land_registry["saon"].str.lower().str.strip()
land_registry["date_of_transfer"] = pd.to_datetime(land_registry["date_of_transfer"])
land_registry_matches = []
for _, match in tqdm(matched_addresses.iterrows(), total=len(matched_addresses)):
# Filter land registry on the postcode
lr_filtered = land_registry[
(land_registry["postcode"] == match["epc_postcode"].lower().strip())
]
# Filter further, when the street is in in the address
# street should be contained in epc_address
lr_filtered = lr_filtered[
lr_filtered["street"].apply(lambda x: is_substring(x, match["epc_address"].lower())) |
lr_filtered["street"].apply(lambda x: is_substring(x, match["Property Address"].lower()))
]
if lr_filtered.empty:
continue
# We now check if paon is in address 1
lr_filtered["paon_match"] = lr_filtered["paon"].apply(lambda x: house_number_match(x, match["house_number"]))
# We also try the secondary match
lr_filtered["saon_match"] = (
lr_filtered["saon"].apply(
lambda x: False if pd.isnull(x) else is_substring(x, match["epc_address1"])
)
)
# We fileter where we have a primary or secondary match
lr_filtered = lr_filtered[
lr_filtered["paon_match"] | lr_filtered["saon_match"]
]
if lr_filtered.empty:
continue
elif lr_filtered.shape[0] == 1:
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
elif lr_filtered.shape[0] > 1:
# We make sure all records are the same and take the newest
all_paon_equal, all_saon_equal, all_street_equal = check_equalities(lr_filtered)
has_paon_match = any(lr_filtered["paon_match"])
if all_paon_equal and all_street_equal and all_saon_equal:
# Take the newest record, append and continue
lr_filtered = lr_filtered.sort_values("date_of_transfer", ascending=False)
lr_filtered = lr_filtered.head(1)
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
elif has_paon_match and all_street_equal:
# Peform filter on paon
lr_filtered = lr_filtered[lr_filtered["paon_match"]]
# Do an addtiioanl equality check
all_paon_equal, all_saon_equal, all_street_equal = check_equalities(lr_filtered)
if all_paon_equal and all_street_equal and all_saon_equal:
lr_filtered = lr_filtered.sort_values("date_of_transfer", ascending=False)
lr_filtered = lr_filtered.head(1)
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
else:
# We do a match on saon
lr_filtered["saon_match2"] = lr_filtered["saon"].apply(
lambda x: False if pd.isnull(x) else is_substring(x, match["epc_address"])
)
lr_filtered = lr_filtered[lr_filtered["saon_match2"]]
if lr_filtered.empty:
continue
elif lr_filtered.shape[0] == 1:
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
else:
raise NotImplementedError("wtf")
else:
# We have a final check, based on an observed case
lr_address_1 = " ".join([x.lower().strip() for x in match["Property Address"].split(",")[0:2]])
lr_filtered["paon_match2"] = lr_filtered["paon"].apply(
lambda x: False if pd.isnull(x) else is_substring(x, lr_address_1)
)
lr_filtered = lr_filtered[lr_filtered["paon_match2"]]
if lr_filtered.empty:
continue
elif lr_filtered.shape[0] == 1:
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
else:
# Check all the same
all_paon_equal, all_saon_equal, all_street_equal = check_equalities(lr_filtered)
# Check saon is house number with exact match
lr_filtered["saon_match2"] = lr_filtered["saon"].apply(
lambda x: False if pd.isnull(x) else house_number_match(x, match["house_number"])
)
# We check if we have a flat
match_flat_number = re.match("flat (\d+)", match["epc_address1"].lower())
match_apartment_number = re.match("apartment (\d+)", match["epc_address1"].lower())
lr_filtered["saon_match3"] = False
if match_flat_number is not None:
# Get out the match
match_flat_number = "flat " + match_flat_number.group(1)
lr_filtered["saon_match3"] = lr_filtered["saon"].apply(
lambda x: False if pd.isnull(x) else x == match_flat_number
)
if match_apartment_number is not None:
# Get out the match
match_apartment_number = "apartment " + match_apartment_number.group(1)
lr_filtered["saon_match3"] = lr_filtered["saon"].apply(
lambda x: False if pd.isnull(x) else x == match_apartment_number
)
if all_paon_equal and all_saon_equal and all_street_equal:
# Take the newest record
lr_filtered = lr_filtered.sort_values("date_of_transfer", ascending=False)
lr_filtered = lr_filtered.head(1)
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
elif any(lr_filtered["saon_match2"]):
lr_filtered = lr_filtered[lr_filtered["saon_match2"]]
all_saon_equal, all_paon_equal, all_street_equal = check_equalities(lr_filtered)
if all_paon_equal and all_saon_equal and all_street_equal:
# Filter on the newest record
lr_filtered = lr_filtered.sort_values("date_of_transfer", ascending=False)
lr_filtered = lr_filtered.head(1)
if lr_filtered.shape[0] == 1:
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
elif any(lr_filtered["saon_match3"]):
lr_filtered = lr_filtered[lr_filtered["saon_match3"]]
if lr_filtered.shape[0] == 1:
land_registry_matches.append(
{
"uprn": match["UPRN"],
"transaction_id": lr_filtered['transaction_id'].values[0],
"price": lr_filtered["price"].values[0],
"date_of_transfer": lr_filtered["date_of_transfer"].values[0],
}
)
continue
raise NotImplementedError("wtf")
else:
raise NotImplementedError("What happened here?")
land_registry_matches = pd.DataFrame(land_registry_matches)
# land_registry_matches.to_excel("land_registry_matches.xlsx")
# Check the matches against the addresses
# lr_to_addresses = matched_addresses[
# ["UPRN", "epc_address", "epc_postcode", "Property Address", "Postcode"]
# ].merge(
# land_registry_matches,
# how="inner",
# left_on="UPRN",
# right_on="uprn"
# ).drop(columns=["uprn"]).merge(
# land_registry[["transaction_id", "paon", "saon", "street", "postcode"]],
# how="left", on="transaction_id"
# )
# Merge onto matched addresses
matched_addresses = matched_addresses.merge(
land_registry_matches,
how="left",
left_on="UPRN",
right_on="uprn"
).drop(columns=["uprn"])
# Flag anything that sold in the last year
matched_addresses["sold_recently"] = (
matched_addresses["date_of_transfer"] >= pd.Timestamp.now() - pd.DateOffset(years=1)
)
matched_addresses["sale_lodged_recently"] = (
(pd.to_datetime(matched_addresses["LODGEMENT_DATE"]) >= pd.Timestamp.now() - pd.DateOffset(months=12)) &
(matched_addresses["TRANSACTION_TYPE"].isin(["marketed sale", "non marketed sale"]))
)
# Save this
# matched_addresses.to_excel("combined_aggregate - pre filter 28th July.xlsx", index=False)
# Drop rows on the booleans
matched_addresses = matched_addresses[
~matched_addresses["sold_recently"] &
~matched_addresses["sale_lodged_recently"]
]
# Filter combined_matching_lookup accordingly
combined_matching_lookup = combined_matching_lookup[
combined_matching_lookup["UPRN"].isin(matched_addresses["UPRN"])
]
# shared_freehold_match = pd.DataFrame(shared_freehold_match)
# Strore these files
# freehold_matching_lookup.to_excel("freehold_matching_lookup.xlsx")
# leasehold_matching_lookup.to_excel("leasehold_matching_lookup.xlsx")
# shared_leasehold_match.to_excel("shared_leasehold_match.xlsx")
# shared_freehold_match.to_excel("shared_freehold_match.xlsx")
# read the files
# freehold_matching_lookup = pd.read_excel("freehold_matching_lookup.xlsx")
# leasehold_matching_lookup = pd.read_excel("leasehold_matching_lookup.xlsx")
# shared_leasehold_match = pd.read_excel("shared_leasehold_match.xlsx")
# freehold_aggregate = aggregate_matches(freehold_matching_lookup, company_ownership, properties)
# leasehold_aggregate = aggregate_matches(leasehold_matching_lookup, company_ownership, properties)
combined_aggregate = aggregate_matches(
matching_lookup=combined_matching_lookup,
company_ownership=company_ownership,
properties=properties
)
investment_50m = combined_aggregate[combined_aggregate["cumulative_value"] <= 51_000_000]
investment_50m_properties = matched_addresses[
matched_addresses["Company Registration No. (1)"].isin(investment_50m["Company Registration No. (1)"])
]
portfolio_epc_data_50m = properties[properties["UPRN"].isin(investment_50m_properties["UPRN"])]
# Storing data
# investment_50m_properties.to_excel("investment_50m_properties 28th July.xlsx", index=False)
# Store the EPC data
# portfolio_epc_data_50m.to_excel("portfolio_epc_data_50m 28th July.xlsx", index=False)
# We check if any of these properties are in a conservation area
valuations = pd.read_excel("property value.xlsx")
uprn_filenames = read_dataframe_from_s3_parquet(
bucket_name="retrofit-data-dev", file_key="spatial/filename_meta.parquet"
)
geospatial_data = []
for _, row in tqdm(valuations.iterrows(), total=len(valuations)):
filtered_df = uprn_filenames[
(uprn_filenames["lower"] <= row["UPRN"])
& (uprn_filenames["upper"] >= row["UPRN"])
]
if filtered_df.empty:
raise Exception("No match found")
filename = filtered_df.iloc[0]["filenames"]
spatial_data = read_dataframe_from_s3_parquet(
bucket_name="retrofit-data-dev", file_key=f"spatial/{filename}"
)
spatial = spatial_data[
spatial_data["UPRN"] == row["UPRN"]
][["UPRN", "conservation_status", "is_listed_building", "is_heritage_building"]]
geospatial_data.append(spatial.to_dict("records")[0])
geospatial_data = pd.DataFrame(geospatial_data)
geospatial_data.to_excel("geospatial_data.xlsx", index=False)
def company_aggregation():
company_ownership = pd.read_csv("/Users/khalimconn-kowlessar/Downloads/CCOD_FULL_2024_04.csv")
aggregation = (
company_ownership
.groupby(["Proprietor Name (1)", "Company Registration No. (1)"])
["Property Address"]
.count()
.reset_index(name="Number of Properties")
)
aggregation = aggregation.sort_values("Number of Properties", ascending=False)
aggregation.to_excel("Company ownership aggregation.xlsx")
def extract_price_info(text):
# Use regex to find the relevant price information
match = re.search(r'Estimated price\n\nLow£([\d,]+)k\n\n£([\d,]+)k\n\nHigh£([\d,]+)k', text)
if match:
low_price = int(match.group(1).replace(',', '')) * 1000
est_price = int(match.group(2).replace(',', '')) * 1000
high_price = int(match.group(3).replace(',', '')) * 1000
price_info = {
'Zoopla Valuation': est_price,
'Zoopla Lower Bound': low_price,
'Zoopla Upper Bound': high_price
}
return price_info
return None
def get_valuations(portfolio_epc_data_50m):
# This gets blocked pretty quickly by Zoopla
import requests
import time
from tqdm import tqdm
valuation_data = []
for _, property_data in tqdm(portfolio_epc_data_50m.iterrows(), total=len(portfolio_epc_data_50m)):
uprn = property_data["UPRN"]
response = requests.get(
f"https://r.jina.ai/https://www.zoopla.co.uk/property/uprn/{uprn}/"
)
pricing = extract_price_info(response.text)
valuation_data.append(
{
"UPRN": uprn,
**pricing
}
)
time.sleep(2)
def prepare_anonymised_data():
investment_50m_properties = pd.read_excel("investment_50m_properties 28th May.xlsx", header=0)
investment_epc_data = pd.read_excel("portfolio_epc_data_50m 28th May.xlsx", header=0)
valuations = pd.read_excel("property value.xlsx", header=0)
# Merge these datasets
df = investment_50m_properties.merge(
investment_epc_data[
["UPRN", "PROPERTY_TYPE", "BUILT_FORM", "TOTAL_FLOOR_AREA", "LODGEMENT_DATE", "POSTCODE"]
].rename(
columns={
"PROPERTY_TYPE": "Property Type",
"BUILT_FORM": "Property Archetype",
"TOTAL_FLOOR_AREA": "Total Floor Area",
"LODGEMENT_DATE": "Date EPC Lodged",
"POSTCODE": "Postcode on EPC"
}
),
how="inner",
on="UPRN"
).merge(
valuations.drop(columns=["ADDRESS", "POSTCODE"]).rename(
columns={
"Zoopla Valuation": "Expected Valuation",
"Zoopla Lower Bound": "Valuation - Lower Bound",
"Zoopla Upper Bound": "Valuation - Upper Bound",
}
),
how="inner",
on="UPRN"
).rename(
columns={
"CURRENT_ENERGY_RATING": "Current EPC",
"CURRENT_ENERGY_EFFICIENCY": "Current SAP Score",
"epc_address": "Address on EPC"
}
).drop(
columns=["Title Number", "match_type", "UPRN"]
)
redacted_owner_names = df[["Company Registration No. (1)"]].drop_duplicates()
redacted_owner_names["Owner"] = ["Owner" + str(i) for i in range(1, len(redacted_owner_names) + 1)]
df = df.merge(
redacted_owner_names, how="left", on="Company Registration No. (1)"
)
df = df.drop(columns=["Company Registration No. (1)", "Proprietor Name (1)", "Property Address"])
df = df.sort_values(["Owner", "Date EPC Lodged"], ascending=False)
redacted_index = []
for _, owner_properties in df.groupby("Owner"):
top_50_percent = round(owner_properties.shape[0] / 2 + 0.00001)
indexes = owner_properties.tail(
owner_properties.shape[0] - top_50_percent
).index
redacted_index.extend(indexes.tolist())
import numpy as np
# Redact addresses and postcodes
df["Address on EPC"] = np.where(
df.index.isin(redacted_index),
"Redacted",
df["Address on EPC"]
)
df["Postcode on EPC"] = np.where(
df.index.isin(redacted_index),
"Redacted",
df["Postcode on EPC"]
)
df.to_excel("Property List - 50% redacted.xlsx", index=False)
def adhoc_change_of_portfolio_analysis_july_2024():
"""
This is just some adhoc analysis, which answers some questions which arose upon refreshing the SFR portfolio
in late July 2024
:return:
"""
# Question 1: Which properties in the previous portfolio were in conservation areas or had listed/heritage status?
def answer_q1():
# Data was just stored here:
geospatial_data = pd.read_excel("geospatial_data.xlsx")
special_buildings = geospatial_data[
(geospatial_data["conservation_status"] == 1) |
geospatial_data["is_listed_building"] |
geospatial_data["is_heritage_building"]
]
print(
f"There were {special_buildings.shape[0]} properties in the previous portfolio which were in conservation "
f"areas or had listed/heritage status"
)
print(f"{(special_buildings['conservation_status'] == 1).sum()} were in a conservation area")
print(f"{special_buildings['is_listed_building'].sum()} were listed buildings")
print(f"{special_buildings['is_heritage_building'].sum()} were heritage buildings")
answer_q1()
# Question 2: For each property in the old portfolio, why was it lost?
def answer_q2():
# We read in the previous 50m portfolio
previous_portfolio = pd.read_excel("investment_50m_properties 28th May.xlsx") # 39 owners
new_matched_addresses = pd.read_excel("combined_aggregate - pre filter 28th July.xlsx")
new_portfolio = pd.read_excel("investment_50m_properties 28th July.xlsx") # 69 owners
# dropped units
dropped_units = previous_portfolio[
~previous_portfolio["UPRN"].isin(new_portfolio["UPRN"].values)
]
# Lots of properties are missed out - why
# 1) What was dropped, but was in the matched addresses and therefore was maybe filtered out
dropped_units_matched = dropped_units[
dropped_units["UPRN"].isin(new_matched_addresses["UPRN"])
].copy()
dropped_units_matched = dropped_units_matched.merge(
new_matched_addresses[
["UPRN", 'transaction_id', 'price', 'date_of_transfer', 'sold_recently', 'sale_lodged_recently']
],
how="left", on="UPRN"
)
# 97 units here - how mant were sold
of_which_sold = dropped_units_matched[
dropped_units_matched["sold_recently"]
]
n_sold = of_which_sold.shape[0]
print(f"{n_sold} sold recently ({n_sold / previous_portfolio.shape[0] * 100})%")
of_which_have_sale_epc_but_not_sold = dropped_units_matched[
~dropped_units_matched["sold_recently"] & dropped_units_matched["sale_lodged_recently"]
]
n_with_sale_epc_but_not_yet_sold = of_which_have_sale_epc_but_not_sold.shape[0]
print(
f"{n_with_sale_epc_but_not_yet_sold} have a sale EPC but have not sold yet ("
f"{n_with_sale_epc_but_not_yet_sold / previous_portfolio.shape[0] * 100})%"
)
# What about things that haven't sold or don't look likely to sell
not_sold = dropped_units_matched[
~dropped_units_matched["sold_recently"] & ~dropped_units_matched["sale_lodged_recently"]
]
new_owner_sizes = new_portfolio.groupby(
["Company Registration No. (1)"]
).size().reset_index().rename(columns={0: "Number of Properties"})
new_owner_sizes = new_owner_sizes.sort_values("Number of Properties", ascending=False)
previous_owner_sizes = previous_portfolio.groupby(
["Company Registration No. (1)"]
).size().reset_index().rename(columns={0: "Number of Properties"})
previous_owner_sizes = previous_owner_sizes.sort_values("Number of Properties", ascending=False)
# Let's just confirm that we took in a bigger owner, as we see this unit was still matched
owner_too_small = []
owner_big_enough = []
for _, property in not_sold.iterrows():
owner_reg_id = property["Company Registration No. (1)"]
old_portfolio_owner_size = previous_owner_sizes[
previous_owner_sizes["Company Registration No. (1)"] == owner_reg_id
]
# We make sure that the number of properties is smaller than the new smallest number
if (
old_portfolio_owner_size["Number of Properties"].values[0] >
new_owner_sizes["Number of Properties"].min()
):
owner_big_enough.append(property.to_dict())
continue
owner_too_small.append(property.to_dict())
n_owner_too_small = len(owner_too_small)
owner_big_enough = pd.DataFrame(owner_big_enough)
summary = []
for _, record in owner_big_enough.iterrows():
# Do we have this new owner?
new_owner = new_portfolio[
new_portfolio["Company Registration No. (1)"] == record["Company Registration No. (1)"]
]
if new_owner.empty:
# Why don't we have this new owner
new_owner_data = new_matched_addresses[
new_matched_addresses["Company Registration No. (1)"] == record["Company Registration No. (1)"]
]
new_owner_data_filtered = new_owner_data[
~new_owner_data["sold_recently"] & ~new_owner_data["sale_lodged_recently"]
]
summary.append(
{
"Owner Name": record["Proprietor Name (1)"],
"Owner reg id": record["Company Registration No. (1)"],
"N properties in new portfolio before filtering": new_owner_data.shape[0],
"N properties in new portfolio after filtering": new_owner_data_filtered.shape[0],
}
)
continue
raise Exception("something went wrong")
summary = pd.DataFrame(summary)
not_accounted_for = summary[
(
summary["N properties in new portfolio before filtering"] <
previous_owner_sizes["Number of Properties"].min()
)
]
# We have two owners not accounted for:
# ALLMID LIMITED, 01959058
# CORAL RACING LIMITED, 541600
# What happened to these owners?
new_epc = pd.read_excel("EPC F & G Properties - V2.xlsx")
allmid = previous_portfolio[previous_portfolio["Company Registration No. (1)"] == "01959058"].copy()
# Check if any of the properties are not in the new EPC data
allmid["not_in_new_epc"] = ~allmid["UPRN"].isin(new_epc["UPRN"])
allmid["not_in_matched_pre_filtered"] = ~allmid["UPRN"].isin(new_matched_addresses["UPRN"])
# In the previous portfolio, Allmid had 4 properties and in the re-build, it has just 2. Why?
# Firstly, one of their properties was re-surveyed not at an F/G
# Secondly, one of their properties is no longer owned by them:
# https://www.zoopla.co.uk/property/uprn/100070553074/
# So as an owner, they fell out of the ranking
coral_racing = previous_portfolio[previous_portfolio["Company Registration No. (1)"] == "541600"].copy()
coral_racing["not_in_new_epc"] = ~coral_racing["UPRN"].isin(new_epc["UPRN"])
coral_racing["not_in_matched_pre_filtered"] = ~coral_racing["UPRN"].isin(new_matched_addresses["UPRN"])
# Coral goes down from 4 -> 1 on refresh, so what happened?
# 1) 2 properties had new EPCs and re-scored higher
# 2) 1 property, 85A Market Street, Church Gresley, Swadlincote, DE11 9PN is no longer matched to the ownership
# data, which is correct
# Why were these units lost?
# There's just 1 owner, who is BARHAM PROPERTY LTD
owner_too_big_ids = owner_big_enough["Company Registration No. (1)"].unique()
owner_too_big_names = owner_big_enough["Proprietor Name (1)"].unique()
previous_owner_size = previous_owner_sizes[
previous_owner_sizes["Company Registration No. (1)"].isin(owner_too_big_ids)
]
new_owner_size = new_matched_addresses[
new_matched_addresses["Company Registration No. (1)"].isin(owner_too_big_ids) |
new_matched_addresses["Proprietor Name (1)"].isin(owner_too_big_names)
]
n_unsold = new_owner_size[~new_owner_size["sold_recently"] & ~new_owner_size["sale_lodged_recently"]].shape
# Happy with the justification to this point
assert (
(n_sold + n_with_sale_epc_but_not_yet_sold + n_owner_too_small + len(owner_big_enough)) ==
dropped_units_matched.shape[0]
)
# We now have a list of properties that were lost from the previous iteration to the next that were not matched
dropped_units_unmatched = dropped_units[
~dropped_units["UPRN"].isin(new_matched_addresses["UPRN"])
].copy()
# A few possibilities: They aren't in the EPC data?
new_epc = pd.read_excel("EPC F & G Properties - V2.xlsx")
unmatched_not_in_epc = dropped_units_unmatched[
~dropped_units_unmatched["UPRN"].isin(new_epc["UPRN"])
]
# There are 17 units that have had new EPCs above a G
# Who were the owners? - various, nothing particularly remarkable
(
previous_portfolio[
previous_portfolio["UPRN"].isin(unmatched_not_in_epc["UPRN"])
]["Proprietor Name (1)"].value_counts()
)
# 22 final units to be accounted for...!
unmatched_in_epc = dropped_units_unmatched[
dropped_units_unmatched["UPRN"].isin(new_epc["UPRN"])
]
# Some of them will be due to ownership
# TODO: Read in freehold/leashold data and see how many of these were non-exact matches!
leasehold_matching_lookup = pd.read_excel("leasehold_matching_lookup V2.xlsx")
freehold_matching_lookup = pd.read_excel("freehold_matching_lookup V2.xlsx")
combined_matching_lookup = pd.concat([leasehold_matching_lookup, freehold_matching_lookup])
# THis is 13 matches, all of them approximate
weak_matches = unmatched_in_epc.merge(combined_matching_lookup, how="inner", on="UPRN")
# These have been lost due to ownership updates. This has been checked manually for every unit and there has
# been sale activity for each one, justifying the change in ownership data
remaining_matches = unmatched_in_epc[
~unmatched_in_epc["UPRN"].isin(weak_matches["UPRN"])
]
assert dropped_units.shape[0] == (
(n_sold + n_with_sale_epc_but_not_yet_sold + n_owner_too_small + len(owner_big_enough)) + len(
weak_matches) + unmatched_not_in_epc.shape[0]
)
Reference in a new issue View git blame Copy permalink